Tasks and Data: Beginner

The difference shows up again when a try fails halfway. The commanding task retries by resubmitting one statement, and before it does, it can ask the warehouse for the state of the job the first try submitted. That statement has to be safe to run twice. If the first try's INSERT INTO host_payouts had already committed when the worker died, a plain resubmit adds the 41,872 host totals a second time, and the table holds 83,744 rows with every host listed twice. A later lesson in this track writes the statement so a rerun replaces the day's rows instead of appending them. The carrying task retries by starting the 7.2GB download from the beginning, inside a worker that may be sitting under the same memory ceiling that ended the first try.

The hard-coded date in both tasks is a flaw. The last section of this lesson replaces it with a value each run fills in for itself.

XCom, short for cross-communication, is the channel one task uses to hand a small value to another task in the same run. The value is a row in the xcom table of Airflow's metadata database. Airflow finds that row by the run id and task id it belongs to, together with the DAG id and a key. When sum_payouts finishes, Airflow writes that row; when notify starts, on whatever worker it lands, it reads the row back. The 2 processes share only that row, which is why the value has to be small enough to sit in a database and plain enough to serialize.

In the last line of that DAG function, passing the result of sum_payouts() into notify(...) draws the edge from sum_payouts to notify, the same dependency >> would draw. The same line also moves the value at run time: the integer sum_payouts returns is pushed to the xcom table under the key return_value, and when notify begins its try, Airflow pulls that row from the same run and passes it in as paid_hosts. The function bodies never mention XCom, because the return statement and the argument stand in for the push and the pull. A task that returns None pushes nothing.

ti is the task instance, which Airflow hands to any @task function that declares a ti parameter. xcom_push writes a row under the key you choose, so one task can publish 2 values: the count under return_value and the job id under job_id, which adds a second row per run to the table above. xcom_pull reads by task id and key, and the key defaults to return_value, so the first pull in notify fetches the count. A pull for a key nobody pushed returns None rather than raising, so a missing push surfaces as a None further down. Because nothing is passed as an argument now, the dependency has to be drawn with >>; without it, notify could start first and pull a row that does not exist yet.

Say the 2026-03-01 run succeeded with sum_payouts pushing 41872. A week later someone clears notify alone to resend the message. notify reruns and pulls 41872, the same row, because the row belongs to sum_payouts and that task was not touched. Now say they clear sum_payouts instead. Clearing it deletes its XCom rows along with its state, and notify is cleared with it as a downstream task. When sum_payouts reruns it pushes a fresh count, 41880 if 8 late reservations landed in the meantime, which notify then pulls. In both cases the XCom row exists exactly as long as the task instance that pushed it keeps its state.

The commanding task in the first section called warehouse.submit(...), which means something built that client from a host and port plus a username and password. Writing those into the DAG file puts a production password into a Python file that is committed to git, copied to every machine that parses DAGs, and printed in any traceback that includes the line. A connection moves those facts out of the file and into a named record that the task fetches by name at run time, so the file says warehouse and nothing more.

A connection record has a fixed set of fields. conn_id is the name tasks use and conn_type is the kind of system. The 5 fields host, port, login, password and schema hold what their names say, and extra is a JSON blob for anything they do not cover, such as a timeout or an SSL flag. The record lives in the metadata database, with the password encrypted with the deployment's key before it is stored, so a database backup holds no plain-text credential.

The UI lists records like these under Admin and Connections, with passwords shown as ***, and a record can be typed in there field by field. The command line does the same job: airflow connections add payments_api --conn-type http --conn-host api.payments.internal --conn-port 8443 --conn-password tok_live_8f3a. And an environment variable named AIRFLOW_CONN_PAYMENTS_API on the worker, holding the record as JSON, is read as the connection payments_api, which suits deployments where secrets arrive through the environment, though a record set that way does not show in the UI's list.

A hook is a Python class, shipped with the package for a kind of system, that takes a conn_id, fetches the record, builds a client from its fields, and exposes methods that use it. Every hook does those 3 things, which is what makes one worth more than a hand-built client: the task names the connection and calls a method, the record supplies where to connect and how to log in, and the hook handles the requests. The HttpHook below reaches the payments service; a hook for a warehouse looks the same from the task's side, with run replaced by a method that executes a statement.

When no hook exists for a system, the task fetches the record itself with BaseHook.get_connection and builds the client by hand. The password still never appears in the file, and Airflow still masks it in the task log: any value held in a connection's password field is replaced by *** wherever it would be printed, so a traceback that dumps the client's settings shows password='***'.

A variable is a key and a value stored in the metadata database for the whole deployment, readable from any task in any DAG. It holds a setting that changes on its own timetable, slower than a run and faster than a deploy: the minimum payout below which a host waits for next week, the addresses a report goes to, the number of days of history to rebuild. Variable.get returns the value as a string, and deserialize_json=True parses a JSON value into a dict or a list.

Variables are set in 3 places. The UI has a Variables page under Admin where a key and a value are typed in. From the command line it is airflow variables set payout_min_amount 20, and airflow variables get payout_min_amount reads it back. The worker's environment is the third place: AIRFLOW_VAR_PAYOUT_MIN_AMOUNT is read as the variable payout_min_amount, which suits values that differ by deployment, though a variable set that way is missing from the UI's list. If a key contains password or secret, its value is masked in the UI and in logs, and the same goes for token and api_key. That makes a variable an acceptable home for a lone API key, though a connection is the better one as soon as a host goes with it.

A param is a setting that belongs to one DAG and can differ per run. The file declares each param with a default, and a scheduled run gets the defaults. A run triggered by hand gets whatever the person triggering it fills in. The UI's trigger form asks for each param in its own field, using the type and description the file gave it. region in the DAG below defaults to all, so the nightly run pays every host, and a rerun for one country is a hand trigger with region set to fr.

params arrives in the task the same way ti did, as a parameter the function declares, and holds this run's values as a dict. Param gives each setting a type, so a trigger form that receives dry_run as the word yes is refused before any run is created, and a plain value such as "all" works where no validation is needed. From the command line the same trigger is airflow dags trigger host_payouts --conf '{"region": "fr", "dry_run": true}', and the values pass through the same validation.

On the afternoon of 2026-03-03 an analyst set payout_min_amount to 25 in the UI, and the last row shows the consequence. The 2026-03-02 run's task ran at midnight at the start of the 3rd and read 20; the 2026-03-03 run's task ran at midnight on the 4th and read 25. The variable keeps no history, so the value a run used is in its log and nowhere else. And if the 2026-03-01 run's sum_payouts is cleared on the 5th and reruns, it reads 25, the current value, because Variable.get executes when the try runs with no memory of what an earlier try read. Params behave in 2 different ways on a rerun. The values a person gives at trigger time are stored with the run when it is created, so a cleared task in the fr run still sees fr and true a month later. A scheduled run is given no values, so it stores none, and each try reads the defaults in the DAG file as the file stands then. Change the file's region default from all to us on the 5th, clear the 2026-03-01 run's sum_payouts, and the rerun's log shows us. Which of the 2 a cleared task sees is decided by how its run was started, and the run_id's manual__ or scheduled__ prefix tells you which.

A templated field is an operator argument whose string Airflow treats as a Jinja template: anything inside {{ }} is an expression, evaluated against the run's context, and the result replaces it. The DAG file holds one task with --day {{ ds }}; each run gets a task instance whose command carries its own date. Rendering happens on the worker, at the start of the try, after the task instance exists and before its code executes, which is why a template can reach XComs pushed by upstream tasks of the same run and the current value of a variable.

4 values in that command came from 4 different places, and the template reached all of them with one syntax. ds is the run's own date, the start of the window it covers, as a YYYY-MM-DD string. params.region is this run's param. var.value.payout_min_amount fetches the variable as the task runs, which is why the 2026-03-03 run renders 25. ti.xcom_pull(...) reads the count sum_payouts pushed in this run, so the command line carries a number computed 20 minutes earlier by another task, on another worker.

Only some arguments are templates. Each operator lists the names it renders in a tuple called template_fields, and for BashOperator that tuple holds bash_command and env plus cwd. A {{ ds }} in any other argument, or in task_id, stays as 8 literal characters. In a @task function, strings passed as arguments to its call are rendered, because they travel as the operator's op_kwargs, which is a templated field, while a string written inside the function body is Python source, which the renderer never reads. The next example prints both, and the 2026-03-01 run prints 2026-03-01 {{ ds }}. Declaring a ds=None parameter is the cleaner route to the same value.

The UI keeps the result. Open a task instance and pick Rendered Templates: it shows every templated field with the string the worker produced for that run, which is the fastest check that {{ params.region }} rendered to fr rather than all, or that the XCom pull found a number rather than None.