Data Pipeline Patterns

Pipeline Stages

A small pipeline can name each intermediate result so the data movement stays visible.

staged pipelines Named stages make filtering, transforming, and summarizing easier to inspect.

Pipeline Stages

show_done
pipeline_stages.py
Replay: real traced execution (multi-file project)
tasks = [
    {"title": "docs", "done": True},
    {"title": "tests", "done": False},
    {"title": "deploy", "done": False},
]

show_done = False

selected = []
for task in tasks:
    if show_done or not task["done"]:
        selected.append(task)

titles = []
for task in selected:
    titles.append(task["title"].upper())

print("tasks=" + ",".join(titles))

tasks = [
    {"title": "docs", "done": True},
    {"title": "tests", "done": False},
    {"title": "deploy", "done": False},
]

show_done = True

selected = []
for task in tasks:
    if show_done or not task["done"]:
        selected.append(task)

titles = []
for task in selected:
    titles.append(task["title"].upper())

print("tasks=" + ",".join(titles))

  1. tasks ← [{'title': 'docs', 'done': True}, {'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}]

    1tasks→ [{'title': 'docs', 'done': True}, {'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}] = [2    {"title": "docs", "done": True},3    {"title": "tests", "done": False},4    {"title": "deploy", "done": False},5]67show_done→ False = False  #@show_done=True89selected→ [] = []10for task in tasks:

  2. for task in tasks:

    pass 1 of 3
    9selected = []10for task{'title': 'docs', 'done': True} in tasks[{'title': 'docs', 'done': True}, {'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}]:11    if show_done or not task["done"]:12        selected.append(task)
    All 3 passes — pass 1 is the card above
    passtaskshow_donetask[”done”]selected
    1{'title': 'docs', 'done': True}———
    2{'title': 'tests', 'done': False}FalseFalse[] → [{'title': 'tests', 'done': False}]
    3{'title': 'deploy', 'done': False}FalseFalse[{'title': 'tests', 'done': False}] → [{'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}]

  3. selected ← [{'title': 'tests', 'done': False}]

    pass 1 of 2
    10for task in tasks:11    if show_doneFalse or not task["done"]False:12        selected→ [{'title': 'tests', 'done': False}].append(task{'title': 'tests', 'done': False})

  4. selected ← [{'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}]

    pass 2 of 2
    10for task in tasks:11    if show_doneFalse or not task["done"]False:12        selected→ [{'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}].append(task{'title': 'deploy', 'done': False})

  5. titles ← []

    14titles→ [] = []15for task in selected:

  6. titles ← ['TESTS']

    pass 1 of 2
    14titles = []15for task{'title': 'tests', 'done': False} in selected[{'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}]:16    titles→ ['TESTS'].append(task["title"]tests.upper())

  7. titles ← ['TESTS', 'DEPLOY']

    pass 2 of 2
    14titles = []15for task{'title': 'deploy', 'done': False} in selected[{'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}]:16    titles→ ['TESTS', 'DEPLOY'].append(task["title"]deploy.upper())

  8. print("tasks=" + ",".join(titles))

    18print("tasks=" + ",".join(titles['TESTS', 'DEPLOY']))
    outputtasks=TESTS,DEPLOY

  1. tasks ← [{'title': 'docs', 'done': True}, {'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}]

    1tasks→ [{'title': 'docs', 'done': True}, {'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}] = [2    {"title": "docs", "done": True},3    {"title": "tests", "done": False},4    {"title": "deploy", "done": False},5]67show_done→ True = True89selected→ [] = []10for task in tasks:

  2. for task in tasks:

    pass 1 of 3
    9selected = []10for task{'title': 'docs', 'done': True} in tasks[{'title': 'docs', 'done': True}, {'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}]:11    if show_done or not task["done"]:12        selected.append(task)
    All 3 passes — pass 1 is the card above
    passtask
    1{'title': 'docs', 'done': True}
    2{'title': 'tests', 'done': False}
    3{'title': 'deploy', 'done': False}

  3. selected ← [{'title': 'docs', 'done': True}]

    pass 1 of 3
    10for task in tasks:11    if show_doneTrue or not task["done"]True:12        selected→ [{'title': 'docs', 'done': True}].append(task{'title': 'docs', 'done': True})
    All 3 passes — pass 1 is the card above
    passtask[”done”]taskselected
    1True{'title': 'docs', 'done': True}[] → [{'title': 'docs', 'done': True}]
    2False{'title': 'tests', 'done': False}[{'title': 'docs', 'done': True}] → [{'title': 'docs', 'done': True}, {'title': 'tests', 'done': False}]
    3False{'title': 'deploy', 'done': False}[{'title': 'docs', 'done': True}, {'title': 'tests', 'done': False}] → [{'title': 'docs', 'done': True}, {'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}]

  4. titles ← []

    14titles→ [] = []15for task in selected:

  5. titles ← ['DOCS']

    pass 1 of 3
    14titles = []15for task{'title': 'docs', 'done': True} in selected[{'title': 'docs', 'done': True}, {'title': 'tests', 'done': False}, {'title': 'deploy', 'done': False}]:16    titles→ ['DOCS'].append(task["title"]docs.upper())
    All 3 passes — pass 1 is the card above
    passtasktask[”title”]titles
    1{'title': 'docs', 'done': True}docs[] → ['DOCS']
    2{'title': 'tests', 'done': False}tests['DOCS'] → ['DOCS', 'TESTS']
    3{'title': 'deploy', 'done': False}deploy['DOCS', 'TESTS'] → ['DOCS', 'TESTS', 'DEPLOY']

  6. print("tasks=" + ",".join(titles))

    18print("tasks=" + ",".join(titles['DOCS', 'TESTS', 'DEPLOY']))
    outputtasks=DOCS,TESTS,DEPLOY