Lesson 2 - Operators

Welcome to Operators#

Every task Module 2 wrote used PythonOperator (directly, or wrapped by @task) because every task was, underneath, “run this Python function.” That’s not the only real choice. An operator is Airflow’s word for what kind of work a task does — PythonOperator runs a Python callable, BashOperator runs a shell command, and dozens of others (covered later in this course) talk to specific systems. This lesson puts a real BashOperator next to a real PythonOperator in the same DAG and shows when each is the right tool.

By the end of this lesson, you will be able to:

• Write a BashOperator task that runs a genuine shell command against a real file
• Explain why BashOperator’s stdout becomes an XCom value the same way a PythonOperator’s return value does
• Choose between PythonOperator and BashOperator based on what a task’s job actually is, not habit

One DAG, Two Kinds of Task#

cityflow_operators_demo_dag.py checks that CityFlow’s real 200,000-row taxi sample file exists, counts its lines with a real shell command, then cross-checks that count against pandas’ own count of the same file — the first two tasks are pure shell work, the third genuinely needs a Python library:

# gate: skip
"""CityFlow's real operators demo -- a BashOperator doing genuine shell work
(checking a real file exists, counting its real lines) feeding a
PythonOperator that cross-checks the shell's answer against pandas' own
row count. Module 3 Lesson 2."""
from datetime import datetime

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

DATA_PATH = "/opt/airflow/dags/data/yellow_tripdata_sample.csv"


def cross_check_row_count(**context):
    """Pull the real `wc -l` line count from the BashOperator upstream (via
    XCom) and compare it against pandas' own real row count for the same
    file -- proving the shell command and the Python library agree."""
    import pandas as pd

    ti = context["ti"]
    wc_output = ti.xcom_pull(task_ids="count_lines_with_bash")
    # `wc -l <path>` prints "<count> <path>"
    line_count = int(wc_output.strip().split()[0])

    df = pd.read_csv(DATA_PATH)
    pandas_rows = len(df)
    # wc -l counts every line, including the header row pandas doesn't count as data
    expected_lines = pandas_rows + 1

    print(f"wc -l reported {line_count} lines")
    print(f"pandas read {pandas_rows} data rows (+1 header = {expected_lines})")
    assert line_count == expected_lines, (
        f"line count mismatch: wc -l said {line_count}, expected {expected_lines}"
    )
    print("shell and Python agree on the real row count")


with DAG(
    dag_id="cityflow_operators_demo",
    schedule=None,
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["cityflow", "module-3", "operators"],
) as dag:
    check_file_exists = BashOperator(
        task_id="check_data_file_exists",
        bash_command=f'test -f "{DATA_PATH}" && echo "found: {DATA_PATH}"',
    )

    count_lines_with_bash = BashOperator(
        task_id="count_lines_with_bash",
        bash_command=f'wc -l "{DATA_PATH}"',
    )

    cross_check = PythonOperator(
        task_id="cross_check_row_count",
        python_callable=cross_check_row_count,
    )

    check_file_exists >> count_lines_with_bash >> cross_check

check_data_file_exists and count_lines_with_bash are both BashOperator — for “does this file exist” and “how many lines does it have,” a shell command is the natural tool, no Python needed. cross_check_row_count has to be PythonOperator: comparing a shell line count against pandas’ own parsed row count requires actually importing pandas, which no shell command can do. Picking the operator is really answering one question: does this task’s job require a Python library, or is it something the shell already does well?

A detail worth naming: BashOperator pushes its command’s stdout to XCom exactly like PythonOperator pushes a return value — that’s how cross_check_row_count can call ti.xcom_pull(task_ids="count_lines_with_bash") and get back the literal text wc -l printed, no different from pulling a Python function’s return value.

Triggered for Real#

import subprocess
import time
import json

dag_id = "cityflow_operators_demo"

deadline = time.time() + 320
found = False
while time.time() < deadline:
    result = subprocess.run(
        ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "list", "--output", "json"],
        capture_output=True, text=True, check=True,
    )
    if any(d["dag_id"] == dag_id for d in json.loads(result.stdout)):
        found = True
        break
    time.sleep(5)
assert found, f"{dag_id} was never picked up by the scheduler"

subprocess.run(
    ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "unpause", dag_id],
    capture_output=True, text=True, check=True,
)
run_id = f"lesson2_verify_{int(time.time())}"
subprocess.run(
    ["docker", "exec", "airflow-local-airflow-scheduler-1",
     "airflow", "dags", "trigger", dag_id, "--run-id", run_id],
    capture_output=True, text=True, check=True,
)
print(f"triggered {dag_id!r} as {run_id!r}")

deadline = time.time() + 60
state = "queued"
while state not in ("success", "failed") and time.time() < deadline:
    time.sleep(3)
    result = subprocess.run(
        ["docker", "exec", "airflow-local-airflow-scheduler-1",
         "airflow", "dags", "list-runs", "-d", dag_id, "--output", "json"],
        capture_output=True, text=True, check=True,
    )
    runs = json.loads(result.stdout)
    state = next(r["state"] for r in runs if r["run_id"] == run_id)

print(f"run {run_id!r} reached state: {state}")
assert state == "success", f"expected success, got {state}"

states = subprocess.run(
    ["docker", "exec", "airflow-local-airflow-scheduler-1",
     "airflow", "tasks", "states-for-dag-run", dag_id, run_id],
    capture_output=True, text=True, check=True,
)
print(states.stdout)

triggered 'cityflow_operators_demo' as 'lesson2_verify_1784885263'
run 'lesson2_verify_1784885263' reached state: success
dag_id                  | task_id                | state
========================+========================+========
cityflow_operators_demo | check_data_file_exists | success
cityflow_operators_demo | count_lines_with_bash  | success
cityflow_operators_demo | cross_check_row_count  | success

All three tasks succeeded, in about three seconds. The real logs show exactly what each operator actually did:

# check_data_file_exists (BashOperator)
Running command: ['/usr/bin/bash', '-c', 'test -f "/opt/airflow/dags/data/yellow_tripdata_sample.csv" && echo "found: ..."']
Output:
found: /opt/airflow/dags/data/yellow_tripdata_sample.csv

# count_lines_with_bash (BashOperator)
Running command: ['/usr/bin/bash', '-c', 'wc -l "/opt/airflow/dags/data/yellow_tripdata_sample.csv"']
Output:
200001 /opt/airflow/dags/data/yellow_tripdata_sample.csv

# cross_check_row_count (PythonOperator)
wc -l reported 200001 lines
pandas read 200000 data rows (+1 header = 200001)
shell and Python agree on the real row count

200,001 real lines in the file, 200,000 real data rows once the header is excluded — the shell’s answer and pandas’ answer agree, because they’re reading the exact same real file two different ways.

Key Takeaways#

• BashOperator and PythonOperator both push their result to XCom the same way — a shell command’s stdout, or a Python function’s return value — so downstream tasks read either kind of upstream result identically via ti.xcom_pull.
• Pick BashOperator when a task’s job is already something the shell does well (check a file, count lines, run an existing CLI tool); pick PythonOperator the moment a task needs a real library like pandas.
• The real numbers agreed because both operators read the identical file: wc -l’s 200,001 lines minus pandas’ 1 header row equals pandas’ own 200,000-row count.
• Airflow’s log SecretsMasker is a real, always-on safety feature, not a display bug — it redacts any literal string matching a configured secret (here, the webserver password airflow) wherever it appears in logs, including inside unrelated file paths.

Exercises#

Exercise 1 — Add a second BashOperator task, check_zone_lookup_exists, that checks taxi_zone_lookup.csv the same way check_data_file_exists does, with no dependency on the existing three tasks. Trigger the DAG and confirm in the Graph view that it runs independently of the others.

Exercise 2 — Change count_lines_with_bash’s command to wc -l "{DATA_PATH}" | awk '{{print $1}}' so it prints only the number, no filename. Update cross_check_row_count to parse the simpler output and confirm the DAG still succeeds with the same real numbers.

Exercise 3 — Write a BashOperator task whose command deliberately references a file that doesn’t exist (e.g. wc -l /opt/airflow/dags/data/nonexistent.csv). Trigger the DAG and use airflow tasks states-for-dag-run to confirm the task reaches failed — and explain, from BashOperator’s documented behavior, why a non-zero exit code from the shell command is what Airflow treats as task failure.