Run it twice — SQL Course

What you'll learn

to assemble a real DAG: @dag, @task, and a dependency defined by a function call
to read a run's protocol: the order of tasks, their states, what landed in the warehouse
to reproduce the failure from ticket #01 yourself: one and the same day, loaded twice
to make a step idempotent by rewriting a partition, and to prove it with a check rather than a promise

Ticket #01, part three. "Make sure it does not happen again"

15:20. The daily report has been recomputed by hand, the Academy has calmed down. A last line appears in the ticket, from the archive warden:

"Good. Now make sure the night-time retry stops doubling the day. The belt is yours."

The report is already fixed. But the work is not over: you have to change the code so that the failure stops being possible.

For that, reproduce it at home first. Not as a separate function launched by hand, but as a small real belt — with tasks, dependencies and a repeated run.

A belt is launched by an orchestrator. It knows the schedule, the order of the tasks and the rules for re-runs. You have already read its records today: the run log, with its attempts and its retries, is its own bookkeeping. Now you will build its belt in code.

We will use Airflow: it comes up often in interviews, and this lesson needs its modern interface.

In an Airflow 3 DAG file, the import looks like this:

from airflow.sdk import dag, task

In our cells it is from arena_airflow import dag, task, run_dag instead. There is one reason: real Airflow cannot be stood up in a browser — it needs a database, a scheduler and a web server.

The decorators and task dependencies below work the same way as they do in Airflow. When you move a DAG like this into an actual dags/, the import is exactly what you replace.

Now the two main entities.

A task is an ordinary function under the @task decorator. A DAG is a function under @dag, inside which tasks are CALLED. Airflow builds the dependencies out of those calls.

The line load(extract()) means: "load comes after extract and receives its result". No separate arrow is needed here — passing the data already sets the order.

An explicit arrow first >> second is needed in the other case: there is an order between the tasks but no data passing. For example: "count the checks first, then publish the mart".

One concept is left, and without it you cannot re-run past dates correctly: logical_date.

It is the DATE WE ARE COUNTING FOR, not the moment of the actual launch. A run for 12 March may happen on the night of the 13th, or a week later while a failure is being investigated. Both times it is obliged to count the same thing.

Now let's assemble a belt of two tasks and run it for one date twice — exactly what the retry did in the night.

A console with a single button, below it on the belt two identical boxes out of one hatch, and to the side an untouched lever marked with a pictogram: take everything out and put it back again. — A run that cannot be repeated without consequences is the one you will repeat — at night, by hand, without thinking.

A two-step belt: extract takes the window from the port, load puts the rows into the warehouse. After the two runs compare not only the states of the tasks but the data as well. You do not need to unpack the port and teaching warehouse mechanics here; focus on the run results. The tasks have the same states, but the warehouse holds twice as many rows. So a green DAG and correct data are not the same thing.

from datetime import datetime
from arena_airflow import dag, task, run_dag
from arena_source import db, fake_api

db.reset()
db.create_table('stg_orders', key=('order_id',), partition_by='day')


@dag(dag_id='nightly_stg', schedule='@daily', start_date=datetime(2184, 3, 1), catchup=False)
def nightly():
    @task
    def extract(**ctx):
        page = fake_api.get('/orders', {'since': ctx['ds'], 'limit': 40})
        return {'day': ctx['ds'], 'rows': page['rows']}

    @task
    def load(payload):
        rows = [{'order_id': r['order_id'], 'day': payload['day']} for r in payload['rows']]
        db.insert('stg_orders', rows)
        return len(rows)

    load(extract())


d = nightly()
print('граф (таска ← от кого зависит):')
print(d.structure())
print()

first = run_dag(d, logical_date=datetime(2184, 3, 12))
print(first.summary())
print('строк на складе после первого прогона :', db.count('stg_orders'))
print()

second = run_dag(d, logical_date=datetime(2184, 3, 12))
print(second.summary())
print('строк на складе после ВТОРОГО прогона :', db.count('stg_orders'))
print('партиции склада                       :', db.partitions('stg_orders'))

from datetime import datetime
from arena_airflow import dag, task, run_dag
from arena_source import db, fake_api

db.reset()
db.create_table('stg_orders', key=('order_id',), partition_by='day')

@dag(dag_id='nightly_stg', schedule='@daily', start_date=datetime(2184, 3, 1), catchup=False)
def nightly():
    @task
    def extract(**ctx):
        page = fake_api.get('/orders', {'since': ctx['ds'], 'limit': 40})
        return {'day': ctx['ds'], 'rows': page['rows']}

@task
    def load(payload):
        rows = [{'order_id': r['order_id'], 'day': payload['day']} for r in payload['rows']]
        db.insert('stg_orders', rows)
        return len(rows)

load(extract())

d = nightly()
print('граф (таска ← от кого зависит):')
print(d.structure())
print()

first = run_dag(d, logical_date=datetime(2184, 3, 12))
print(first.summary())
print('строк на складе после первого прогона :', db.count('stg_orders'))
print()

second = run_dag(d, logical_date=datetime(2184, 3, 12))
print(second.summary())
print('строк на складе после ВТОРОГО прогона :', db.count('stg_orders'))
print('партиции склада                       :', db.partitions('stg_orders'))

python · airflow · источник

A bare INSERT is not a load

Forty rows turned into eighty. And not one task fell over: both DAG runs finished successfully.

That is exactly what happened on the night of 13 March. Only there an automatic retry stood in for your button, and a day's revenue stood in for the forty rows.

QUERY: A green run means the step reached the end. Not that it did the right thing. Noticing the difference between those two statements is what you are paid for.

Why did it happen? The cause is one line: db.insert(...) appends data to what is already lying in the warehouse.

For an event that is guaranteed to happen once, that is a perfectly normal operation. But a load for a date is not such an event. One and the same day is loaded as many times as the step is launched.

Repeated launches are unavoidable: orchestrators can automatically rerun tasks after a failure. In Airflow, retries can be enabled with default_args={'retries': 2}. By default there are NO retries: a task has retries = 0 until they are configured.

But having no automatic retry does not solve the problem. Whatever the orchestrator does not re-run at night, the person going through the mess in the morning will.

So what we need is a different property of the step — idempotency. However many times the step is executed with the same inputs, the resulting state must stay the same.

Note: idempotency does not mean "a repeated run will not fall over". It means that the state of the warehouse after the second run is indistinguishable from the state after the first.

To get that property, change the unit of writing. The loader has to think not in individual rows but in a partition — a slice of data belonging wholly to one date.

Every launch then does two things:

take off everything belonging to this date — delete_partition(table, day);
put back what has just been counted — insert(table, rows).

Look at what that changes. The first run deletes zero rows and puts down forty. The second deletes forty and puts down forty. The third does the same.

The result after every launch is identical. And neighbouring dates are not touched: only ITS OWN partition is deleted, not the whole table.

The cure from de1l1 makes more sense now, too. The doubled day could not be fixed by a plain re-run: a step that only appends would have put down a third set of rows. What was needed was exactly a rewrite of the range.

What is left is not to call the step idempotent but to prove it.

A comment saying "this step is idempotent" guarantees nothing — that is the very comment that hung over your predecessor's loader. The check has to reproduce a repeated launch and compare the state before and after it.

So the proof here is simple: two runs in a row and a comparison of warehouse snapshots. A is db.snapshot(table), a SHA-256 hash of the contents. If the snapshots before and after the repeat match, the repeated run really did not change the warehouse state.

Three runs of one date by three ways of writing: a bare INSERT gives 40 → 80 → 120 rows, rewriting its own partition gives 40 → 40 → 40, and truncate gives the same total and wipes the neighbouring day.

What is real here and what is sandbox. arena_airflow reproduces the PUBLIC interface of Airflow 3: DAG, @dag, @task, PythonOperator, the >> arrows, XCom, TaskGroup, sensors, logical_date, ds, retries and retry_delay. You do not need to unpack the whole list yet: this lesson focuses on decorators, dependencies and the run date.

That part carries over into a real dags/: it is enough to replace the import line with from airflow.sdk import dag, task.

But around the DAG the sandbox has teaching scaffolding that does NOT exist in Airflow. On the way over it is thrown away wholesale: run_dag(...), d.structure(), run.summary() and the warehouse arena_source.db (create_table, insert, delete_partition, snapshot, partitions).

In their role is played by the airflow dags test command, the web interface and your real database. What carries over is the body of the tasks, not the console we launch them from in the lesson.

There are larger simplifications too. The shim has no scheduler, no database, no web interface, no pools and slots, no executors (Celery, Kubernetes), no deferred tasks with a triggerer, no detection of hung processes and no real cron — and never will.

The schedule here is computed from dates, not from the server's clock. Tasks run strictly one at a time, in dependency order: a parallel executor would give different output on different runs. Retries do not really sleep either — retry_delay accumulates on the run's virtual clock.

So parallelism, resources and queues are things this course asks you to take on trust. But DAG semantics, ordering, retries, idempotency and backfill — running a sequence of past dates — are reproduced faithfully here. Those are exactly the topics that come up in interviews.

Practice: write the code

Fix the belt so that re-running one and the same date does not change the state of the warehouse. The starter holds the same DAG as the example above. Edit only the body of the load task. Three conditions have to hold after the repair:

two runs for 12 March in a row leave the same number of rows in the warehouse and the same db.snapshot('stg_orders');
a run for 13 March ADDS its own partition and does not touch the partition of the 12th;
all runs stay green.

For that the warehouse gives you db.delete_partition(table, partition): the function removes one partition and returns the number of rows deleted. The partitioning key is declared as partition_by='day', and the date you need sits in payload['day'].

from datetime import datetime
from arena_airflow import dag, task
from arena_source import db, fake_api


def make_dag():
    """Собирает и возвращает ночной DAG. Правь только тело таски load."""

    @dag(dag_id='nightly_stg', schedule='@daily', start_date=datetime(2184, 3, 1), catchup=False)
    def nightly():
        @task
        def extract(**ctx):
            page = fake_api.get('/orders', {'since': ctx['ds'], 'limit': 40})
            return {'day': ctx['ds'], 'rows': page['rows']}

        @task
        def load(payload):
            rows = [{'order_id': r['order_id'], 'day': payload['day']}
                    for r in payload['rows']]
            db.insert('stg_orders', rows)
            return len(rows)

        load(extract())

    return nightly()

python · airflow · источник

Interview question

How this is asked at interview. "What is an idempotent load, and how do you make a load idempotent?"

Start with a definition through the REPEAT: after any number of runs with the same inputs, the result is the same as after one.

After the definition they want the mechanics, and here you need specifics: rewriting a partition (DELETE of a range + INSERT, or INSERT OVERWRITE, or MERGE on a key). You must rewrite the partition selected by the run's logical date.

An extra mark goes for the phrase "the load works by logical_date, not by 'today'". Otherwise a retry for last week takes the current date and counts an entirely different set of data.

The next question is usually "how do you prove idempotency?" The right answer is with a test that runs the step twice and compares the resulting state. A comment in the code proves nothing.

Another frequent one: "and what if the task fell over AFTER writing but before the success mark?" That case is exactly what idempotency is for. The orchestrator starts the retry from the beginning of the step, so the repeat has to bring the warehouse to the same state.

Check yourself

The load task fell over AFTER writing rows to the warehouse but before the success mark. The orchestrator launches a retry. What happens if the step can only append rows?

Check yourself

A colleague made load idempotent like this: before inserting, it calls db.truncate('stg_orders'). Two runs for 12 March give the same result, but the load must preserve other dates. What is wrong?

Key takeaways

The DAG is assembled: @dag wraps the function, @task marks a single step, and the dependency arises from the call load(extract()). The >> arrow is for when there is an order but no data passing.
logical_date is the date we are counting FOR, not the moment of launch. A run for 12 March must give the same result on the night of the 13th and a week later.
A bare INSERT is not a load: two green runs for one date gave 80 rows instead of 40. That is how the example reproduces the failure of ticket #01.
The cure is rewriting ITS OWN partition: delete_partition(table, day) before insert. Not truncate: that wipes out neighbouring dates and breaks the very first backfill.
Idempotency has to be proved: two runs in a row and a comparison of db.snapshot(). A comment in the code does not count as proof.

Further down the belt is the chapter's last lesson: what goes into the watch log, and which queries tell you in the morning that the night went normally.