Side Effects Inside Transformations

A function runs each time its task runs

A function passed to a transformation runs on an executor, inside a task, every time that task executes. Everything earlier in this lesson multiplies task executions: each action replays the plan, a reliable save makes 2 passes, a lost executor's stages get resubmitted, a failed task is retried, and a slow one may get a speculative copy. A function that only computes a value from its row is safe through all of it. A function that also changes something outside the row, such as a counter, a log, an API call or another database, makes that change once per execution.

The counter that reads 2,000

Accumulators are Spark's shared counters: tasks add to them, and the driver reads the total. Here a validation function counts payments with no amount as the rows pass through map, and 2 actions follow.

missing_amount = spark.sparkContext.accumulator(0)
payments = spark.table("payments")

def validate(row):
    if row.amount is None:
        missing_amount.add(1)
    return row

checked = payments.rdd.map(validate)

print(checked.count())
spark.createDataFrame(checked, payments.schema).write.parquet("/warehouse/payments_checked/")
print(missing_amount.value)

Say 1,000 payments have no amount. count() runs validate over every row and the counter reaches 1,000. The write is a second action, so it reads the table and runs validate again, and the driver prints 2,000. Spark's documentation states the rule: an accumulator update made inside an action is applied once per task, and an update made inside a transformation may be applied more than once if tasks or stages are re-executed.

Count the ways a row's function can run, and the count of its effect stops being 1:

Effects go in actions, with a key

Move the effect out of the transformation and into an action. foreachPartition is an action that hands your function an iterator over a partition's rows, so the function opens a connection per partition, sends what it needs to, and closes it. Accumulator updates made inside it count once per task, retries included.

def send_alerts(rows):
    client = AlertClient()
    for row in rows:
        client.upsert(alert_id=row.txn_id, body=row.reason)
    client.close()

flagged = (
    spark.table("payments")
    .filter(F.col("amount").isNull())
    .select("txn_id", F.lit("missing amount").alias("reason"))
)
flagged.foreachPartition(send_alerts)

An action still runs again under retries and speculation, and an external effect doesn't roll back. A task that delivered 400 of its 500 alerts and then failed delivers all 500 again on its retry. So the receiver has to tolerate repeats, and that needs a stable key taken from the row, here alert_id=row.txn_id. Designing writes that land once on such a key is the subject of the fault-tolerance lessons later in the track.

Do

• Compute values in transformations; send, write and count in foreachPartition or a write
• Give every external write an id from the row, so a repeat overwrites the earlier one
• Read an accumulator after the action that fills it, and fill it only inside actions
• Open a connection per partition inside foreachPartition, never one per row

Avoid

• Call an API, send a message or write a file from a function passed to map
• Report an accumulator filled inside a transformation as an exact count
• Debug with print in a transformation: the output goes to executor logs
• Assume a reliable checkpoint() runs your function once: the save makes 2 passes

push_notifs holds 1,000 rows in 4 partitions, and send_push delivers a notification per call. After both actions run, which is true?

sent = spark.sparkContext.accumulator(0)

def notify(row):
    send_push(row.user_id)
    sent.add(1)
    return row

pushed = spark.table("push_notifs").rdd.map(notify)
pushed.count()
pushed.take(5)

• 1,000 pushes were sent and sent reads 1,000, because take(5) reuses the rows count() computed
• 1,005 pushes were sent and sent reads 1,005, because take(5) reran notify on 5 rows
• 2,000 pushes were sent and sent reads 2,000, because each action replays all 4 partitions
• 1,005 pushes were sent but sent reads 1,000, because accumulators skip a second action's updates

About this lesson

This section is part of the Lazy Evaluation: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.