Apache Spark Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Apache Spark and its hosted variants (Databricks, Amazon EMR, Google Dataproc) are increasingly the execution layer for large-scale AI agent workloads — batch inference pipelines that process millions of records, multi-agent fan-out jobs that call external APIs per-record, and Structured Streaming applications that react to events in near-real time. When those pipelines include billing logic — calling stripe.charges.create() per customer record, per usage event, or per inference result — Spark's distributed execution model introduces three specific failure modes that neither Stripe restricted keys nor standard idempotency discipline fully prevents without understanding how Spark tasks actually execute.

This post covers all three failure modes with PySpark code, and the two-layer governance pattern — content-hash idempotency keys plus per-partition vault keys via a spend-cap proxy — that eliminates all three without changing your cluster configuration or Spark version.

Failure mode 1: task retry on executor failure re-executes stripe.charges.create() on the new executor

Spark's fault-tolerance model is built on task re-execution. When an executor fails — due to out-of-memory errors, spot instance preemption, hardware failure, or network timeout — Spark marks the tasks that were running on that executor as failed and re-schedules them on other available executors. The re-scheduled tasks start from the beginning of the task function and execute all the way through. For pure data transformations, this is safe: reading the same partition, applying the same function, and writing the same output is idempotent. For tasks that call external APIs, it is not.

A PySpark foreachPartition() function that calls stripe.charges.create() per record will, on executor failure, re-execute every Stripe call for every record in the partition — including records that were already successfully charged before the executor failed. Spark provides no mechanism to communicate partial completion state to the retry: the new executor starts the partition function from record zero with no knowledge of which charges already succeeded.

# billing.py — UNSAFE: no idempotency key
from pyspark.sql import SparkSession
import stripe
import os

def charge_partition(rows):
    """Called once per partition. Each row is a customer billing record."""
    stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

    for row in rows:
        # UNSAFE: If this executor OOMs after charging row 47 and before
        # completing the partition, Spark retries the entire partition on
        # a new executor starting from row 0. Rows 0-46 are charged again.
        # Stripe creates ch_B, ch_C... for each already-charged customer.
        result = stripe.charges.create(
            amount=row["amount_cents"],
            currency="usd",
            customer=row["customer_id"],
            description=f"Billing {row['billing_period']}",
        )
        yield (row["customer_id"], result["id"], "charged")

spark = SparkSession.builder.appName("billing").getOrCreate()
customers_df = spark.read.parquet("s3://data/customers-2026-07/")
customers_rdd = customers_df.rdd

results_rdd = customers_rdd.mapPartitions(charge_partition)
results_rdd.toDF(["customer_id", "charge_id", "status"]).write.parquet("s3://data/billing-results/")

The same risk applies to speculative execution. When spark.speculation is true (the default on many managed clusters including Databricks), Spark launches a duplicate copy of any task that is running significantly slower than the median for that stage. Both the original and the speculative copy call stripe.charges.create() for the same records in parallel. The first copy to complete wins, but both may have already fired Stripe API calls — creating duplicate charges without an idempotency key.

The fix is a content-hash idempotency key derived deterministically from each record's immutable billing fields. Because the idempotency key is computed from the record itself — not from a task ID, executor ID, or attempt counter that changes between retries — every retry of the task function sends the same key to Stripe for each record. Stripe returns the original charge object on the retry rather than creating a new charge.

# billing.py — SAFE: content-hash idempotency key per record
from pyspark.sql import SparkSession
import stripe
import hashlib
import os

def billing_idempotency_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    raw = f"{customer_id}:{amount_cents}:{billing_period}:spark-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def charge_partition(rows):
    """Per-partition billing. Idempotency key is stable across task retries."""
    vault_key = os.environ.get("KEYBRAKE_VAULT_KEY") or os.environ["STRIPE_SECRET_KEY"]
    stripe_client = stripe.Stripe(
        api_key=vault_key,
        base_url="https://proxy.keybrake.com/stripe",
    )

    for row in rows:
        idempotency_key = billing_idempotency_key(
            row["customer_id"], row["amount_cents"], row["billing_period"]
        )
        try:
            result = stripe_client.charges.create(
                params={
                    "amount": row["amount_cents"],
                    "currency": "usd",
                    "customer": row["customer_id"],
                    "description": f"Billing {row['billing_period']}",
                },
                options={"idempotency_key": idempotency_key},
            )
            yield (row["customer_id"], result.id, "charged")
        except stripe.error.CardError as e:
            yield (row["customer_id"], None, f"card_declined:{e.user_message}")
        except stripe.error.InvalidRequestError as e:
            yield (row["customer_id"], None, f"invalid_request:{e.user_message}")

spark = SparkSession.builder.appName("billing").getOrCreate()
customers_df = spark.read.parquet("s3://data/customers-2026-07/")

results_rdd = customers_df.rdd.mapPartitions(charge_partition)
results_rdd.toDF(["customer_id", "charge_id", "status"]).write.parquet("s3://data/billing-results/")

Note that permanent Stripe errors (card declined, invalid customer, invalid request) are caught and yielded as structured error records rather than re-raised. Re-raising a permanent error inside mapPartitions() causes the task to fail, which triggers Spark's retry logic — which then retries the entire partition including all the records that succeeded before the error. Catch permanent errors inline and emit them as output records so the task completes successfully and Spark does not retry the partition.

Failure mode 2: foreachPartition() shares one unrestricted STRIPE_SECRET_KEY across all concurrent executors

In PySpark, environment variables set on the driver are not automatically propagated to executor processes. The standard pattern is to pass the Stripe key via a Spark broadcast variable, via executor environment configuration (spark.executorEnv.STRIPE_SECRET_KEY), or by serializing the key into the partition function's closure. All three methods result in the same key being available on every executor simultaneously. If you run a billing job with 200 partitions across 50 executors, 200 concurrent Stripe calls all use the same unrestricted STRIPE_SECRET_KEY with no per-partition or per-executor dollar cap.

The practical risk is a silent data error in amount_cents. If a transformation upstream of the billing stage produces wrong amounts — off by a factor of 100 due to a currency unit confusion, or missing a decimal conversion step — all 200 partitions begin charging customers the wrong amount simultaneously. There is no circuit breaker. By the time any individual charge returns a result and you detect the error, the majority of the fan-out has already fired. A $49.99/month subscription incorrectly charged at $4,999 to 10,000 customers is a $49.99M error.

# UNSAFE: one STRIPE_SECRET_KEY shared across all 200 concurrent partitions
# spark-submit --conf spark.executorEnv.STRIPE_SECRET_KEY=sk_live_... billing.py
#
# If amount_cents contains a data error, all 200 partitions fire simultaneously.
# No per-partition cap. No circuit breaker.

def charge_partition_unsafe(rows):
    stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # same key on every executor
    for row in rows:
        stripe.charges.create(amount=row["amount_cents"], ...)
        yield ...

The fix is per-partition vault keys. Before dispatching the billing job, the driver issues one vault key per partition from the Keybrake proxy, each capped at that partition's total expected charge amount plus a small buffer (10–15%). The vault keys are broadcast to the executors as a mapping from partition index to key. Each partition function looks up its assigned vault key and uses only that key for all charges in its partition. A data error that inflates amount_cents causes the vault key's cap to be hit after the first overcharged record, blocking the rest of the partition without affecting other partitions.

# billing_safe.py — per-partition vault keys via Keybrake
from pyspark.sql import SparkSession
from pyspark import TaskContext
import stripe
import hashlib
import requests
import os

def issue_vault_key(partition_total_cents: int, billing_period: str) -> str:
    """Issue a vault key capped to this partition's expected total spend."""
    resp = requests.post(
        "https://proxy.keybrake.com/keys",
        json={
            "vendor": "stripe",
            "daily_usd_cap": round(partition_total_cents / 100 * 1.10, 2),
            "allowed_endpoints": ["/v1/charges"],
            "expires_in_seconds": 3600,
            "label": f"spark-billing-{billing_period}",
        },
        headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_MASTER_KEY']}"},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

def charge_partition_safe(rows, vault_key_broadcast):
    """Uses a per-partition vault key. Cap hit = only this partition is blocked."""
    ctx = TaskContext.get()
    partition_id = ctx.partitionId()
    vault_key = vault_key_broadcast.value.get(str(partition_id))

    if not vault_key:
        raise RuntimeError(f"No vault key for partition {partition_id}")

    stripe_client = stripe.Stripe(
        api_key=vault_key,
        base_url="https://proxy.keybrake.com/stripe",
    )

    for row in rows:
        idempotency_key = billing_idempotency_key(
            row["customer_id"], row["amount_cents"], row["billing_period"]
        )
        try:
            result = stripe_client.charges.create(
                params={
                    "amount": row["amount_cents"],
                    "currency": "usd",
                    "customer": row["customer_id"],
                    "description": f"Billing {row['billing_period']}",
                },
                options={"idempotency_key": idempotency_key},
            )
            yield (row["customer_id"], result.id, "charged")
        except stripe.error.CardError as e:
            yield (row["customer_id"], None, f"card_declined:{e.user_message}")
        except stripe.error.InvalidRequestError as e:
            yield (row["customer_id"], None, f"invalid_request:{e.user_message}")

# Driver: compute per-partition totals, issue vault keys, broadcast
spark = SparkSession.builder.appName("billing-safe").getOrCreate()
customers_df = spark.read.parquet("s3://data/customers-2026-07/")

# Compute partition totals on the driver before dispatching
partition_totals = (
    customers_df
    .rdd
    .mapPartitionsWithIndex(lambda idx, rows: [(idx, sum(r["amount_cents"] for r in rows))])
    .collectAsMap()
)

billing_period = "2026-07"
vault_keys = {
    str(pid): issue_vault_key(total, billing_period)
    for pid, total in partition_totals.items()
}
vault_keys_bc = spark.sparkContext.broadcast(vault_keys)

results_rdd = customers_df.rdd.mapPartitions(
    lambda rows: charge_partition_safe(rows, vault_keys_bc)
)
results_rdd.toDF(["customer_id", "charge_id", "status"]).write.parquet("s3://data/billing-results/")

The two-pass approach — first compute partition totals, then issue vault keys, then dispatch billing — adds one extra Spark action before the billing run. This is acceptable overhead for a billing job where the alternative is an uncapped fan-out. Note that vault keys issued before the billing stage must have a TTL long enough to survive the entire billing job runtime. Issue vault keys with a 60-minute TTL for jobs that typically complete in 30 minutes, and 120 minutes for larger jobs.

Failure mode 3: Structured Streaming foreachBatch() checkpoint replay double-bills the same micro-batch

Spark Structured Streaming processes data in micro-batches. After each micro-batch completes, the query engine commits a checkpoint: a record of the source offsets consumed and the output written. Checkpointing enables recovery — if the streaming query fails, Spark restarts from the last committed checkpoint, which means re-processing the micro-batch that was in flight at the time of failure.

The standard pattern for calling external APIs from Structured Streaming is foreachBatch(): a function called once per micro-batch with a DataFrame of the batch's records. If your foreachBatch() function calls stripe.charges.create() for billing events, and the streaming query fails after the Stripe calls complete but before the checkpoint commits — or if the checkpoint commits but the downstream sink write fails and the batch is retried — the same micro-batch is delivered to foreachBatch() again. Without idempotency keys, those Stripe charges fire a second time.

# streaming_billing.py — UNSAFE: foreachBatch without idempotency keys
from pyspark.sql import SparkSession
import stripe
import os

def process_billing_batch(batch_df, batch_id):
    """Called for each micro-batch. UNSAFE: no idempotency key."""
    stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

    # If this function completes Stripe calls but the checkpoint write fails,
    # this function runs again for the same batch_id with the same records.
    # Without idempotency keys, each record is charged twice.
    for row in batch_df.collect():
        stripe.charges.create(
            amount=row["amount_cents"],
            currency="usd",
            customer=row["customer_id"],
            description=f"Billing {row['billing_period']}",
        )

spark = SparkSession.builder.appName("streaming-billing").getOrCreate()

events_df = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "billing-events")
    .load()
)

query = (
    events_df
    .writeStream
    .foreachBatch(process_billing_batch)
    .option("checkpointLocation", "s3://checkpoints/billing/")
    .start()
)
query.awaitTermination()

The fix for streaming billing combines three elements. First, include the batch_id in the idempotency key as a positional component — this makes the key stable across replays of the same batch for the same record. Second, include the record's business fields (customer ID, amount, billing period) in the key so that two different batches that happen to contain the same customer don't share a key. Third, use a per-batch vault key capped at the batch's total expected charge amount, issued at the start of foreachBatch() before any Stripe calls.

# streaming_billing.py — SAFE: batch-scoped idempotency keys + per-batch vault key
from pyspark.sql import SparkSession
import stripe
import hashlib
import requests
import os

KEYBRAKE_MASTER_KEY = os.environ["KEYBRAKE_MASTER_KEY"]

def billing_idempotency_key(customer_id: str, amount_cents: int,
                             billing_period: str, batch_id: int) -> str:
    raw = f"{customer_id}:{amount_cents}:{billing_period}:{batch_id}:spark-streaming-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def issue_batch_vault_key(batch_total_cents: int, batch_id: int) -> str:
    resp = requests.post(
        "https://proxy.keybrake.com/keys",
        json={
            "vendor": "stripe",
            "daily_usd_cap": round(batch_total_cents / 100 * 1.10, 2),
            "allowed_endpoints": ["/v1/charges"],
            "expires_in_seconds": 300,
            "label": f"spark-stream-batch-{batch_id}",
        },
        headers={"Authorization": f"Bearer {KEYBRAKE_MASTER_KEY}"},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

def process_billing_batch(batch_df, batch_id):
    """foreachBatch handler. Idempotency key includes batch_id — stable on replay."""
    rows = batch_df.collect()
    if not rows:
        return

    batch_total_cents = sum(r["amount_cents"] for r in rows)
    vault_key = issue_batch_vault_key(batch_total_cents, batch_id)

    stripe_client = stripe.Stripe(
        api_key=vault_key,
        base_url="https://proxy.keybrake.com/stripe",
    )

    results = []
    for row in rows:
        idempotency_key = billing_idempotency_key(
            row["customer_id"], row["amount_cents"],
            row["billing_period"], batch_id,
        )
        try:
            charge = stripe_client.charges.create(
                params={
                    "amount": row["amount_cents"],
                    "currency": "usd",
                    "customer": row["customer_id"],
                    "description": f"Billing {row['billing_period']}",
                },
                options={"idempotency_key": idempotency_key},
            )
            results.append((row["customer_id"], charge.id, "charged"))
        except stripe.error.CardError as e:
            results.append((row["customer_id"], None, f"card_declined:{e.user_message}"))

    # Write results to sink — this is the operation that must be idempotent
    # (or wrapped in an upsert) to handle batch replay at the sink layer too
    spark = batch_df.sparkSession
    results_df = spark.createDataFrame(results, ["customer_id", "charge_id", "status"])
    results_df.write.mode("append").parquet("s3://data/billing-results/")

spark = SparkSession.builder.appName("streaming-billing-safe").getOrCreate()

events_df = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "billing-events")
    .load()
)

query = (
    events_df
    .writeStream
    .foreachBatch(process_billing_batch)
    .option("checkpointLocation", "s3://checkpoints/billing-safe/")
    .start()
)
query.awaitTermination()

Note that the sink write inside foreachBatch() — the results_df.write.parquet() call — also needs to handle replay correctly. Using mode("append") with Parquet will create duplicate output records on replay. For production streaming billing, use a sink that supports upsert semantics (Delta Lake MERGE, Iceberg MERGE INTO, or a database with an ON CONFLICT DO UPDATE write) so that replayed output rows replace rather than duplicate the originals.

Approach comparison

pytest enforcement suite

# test_spark_billing.py
import hashlib
import pytest

def billing_idempotency_key(customer_id: str, amount_cents: int,
                             billing_period: str) -> str:
    raw = f"{customer_id}:{amount_cents}:{billing_period}:spark-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def streaming_idempotency_key(customer_id: str, amount_cents: int,
                               billing_period: str, batch_id: int) -> str:
    raw = f"{customer_id}:{amount_cents}:{billing_period}:{batch_id}:spark-streaming-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def test_batch_idempotency_key_stable_across_task_retries():
    """Same record → same key regardless of task attempt number."""
    k1 = billing_idempotency_key("cus_abc", 4999, "2026-07")
    k2 = billing_idempotency_key("cus_abc", 4999, "2026-07")
    assert k1 == k2
    assert len(k1) == 32

def test_batch_idempotency_key_differs_across_customers():
    k_a = billing_idempotency_key("cus_abc", 4999, "2026-07")
    k_b = billing_idempotency_key("cus_xyz", 4999, "2026-07")
    assert k_a != k_b

def test_streaming_idempotency_key_stable_across_batch_replay():
    """Same batch_id + same record → same key on checkpoint replay."""
    k1 = streaming_idempotency_key("cus_abc", 4999, "2026-07", batch_id=42)
    k2 = streaming_idempotency_key("cus_abc", 4999, "2026-07", batch_id=42)
    assert k1 == k2

def test_streaming_idempotency_key_differs_across_batches():
    """Different batch_id → different key even for same record."""
    k_b42 = streaming_idempotency_key("cus_abc", 4999, "2026-07", batch_id=42)
    k_b43 = streaming_idempotency_key("cus_abc", 4999, "2026-07", batch_id=43)
    assert k_b42 != k_b43

def test_vault_key_not_bare_stripe_key(monkeypatch):
    monkeypatch.setenv("STRIPE_SECRET_KEY", "sk_live_real_secret")
    monkeypatch.setenv("KEYBRAKE_VAULT_KEY", "kb_vault_test_partition_0")
    import os
    vault_key = os.environ["KEYBRAKE_VAULT_KEY"]
    assert vault_key.startswith("kb_"), "Spark billing must use vault key, not bare Stripe key"
    assert vault_key != os.environ["STRIPE_SECRET_KEY"]

Gap analysis

1. Spark's speculative execution can fire duplicate stripe.charges.create() calls from two concurrent task copies before either completes

When spark.speculation=true (default on Databricks and many managed clusters), Spark launches a speculative copy of any task running more than 1.5× the median task duration for that stage. Both the original and the speculative copy execute the partition function concurrently — calling stripe.charges.create() for the same records at the same time. With a content-hash idempotency key, both calls send the same key to Stripe for each record, and Stripe deduplicates them: the first call to arrive creates the charge, the second returns the original charge object. Without an idempotency key, both calls create separate charges and there is no Spark-level mechanism to detect or prevent this. Idempotency keys are especially important when speculative execution is enabled. You can also disable speculative execution for billing stages specifically by setting spark.speculation=false at the job level or by isolating the billing stage into a separate job that runs with speculation disabled.

2. spark.task.maxFailures controls how many times a task is retried — but the default of 4 means up to 5 total executions of your billing partition function per partition

Spark retries a failed task up to spark.task.maxFailures times (default: 4, meaning 5 total attempts) before marking the stage as failed. For a billing partition function that charges 500 customers, a task that fails consistently (perhaps due to a persistent downstream error that doesn't affect the Stripe calls themselves) will attempt 500 Stripe calls up to 5 times — 2,500 total Stripe API calls for what should be 500 charges. With idempotency keys, all 2,500 calls return the same 500 charge objects. Without idempotency keys, all 2,500 calls create new charges. For billing jobs with consistently failing tasks (disk pressure, executor OOM, data skew causing timeouts), investigate and fix the underlying cause rather than increasing spark.task.maxFailures, which amplifies the blast radius of any billing logic inside the failing tasks.

3. Vault keys issued by the driver before a large billing job may expire during the job if the job takes longer than the vault key TTL

Vault keys issued with a 60-minute TTL will expire if a billing job runs for more than 60 minutes. A partition function that attempts to use an expired vault key will receive a 401 from the proxy, raising a stripe.error.AuthenticationError inside the partition. If this error is caught and yielded as an error record (the recommended pattern for permanent errors), the task completes but those customer records are left uncharged. If it propagates as an exception, the task fails and Spark retries it — but the new task attempt still uses the expired vault key. Issue vault keys with a TTL equal to 150% of the expected job runtime, and implement a fallback in the partition function that re-issues a fresh vault key if it detects a 401 response from the proxy. For very long-running jobs (multi-hour billing runs), consider issuing vault keys at the beginning of each partition execution rather than pre-issuing them on the driver — this requires a call to the Keybrake API inside the partition function but avoids TTL expiry for large fan-outs.

4. mapPartitions() receives a Python iterator — consuming the iterator twice (e.g., to compute a total before charging) re-reads partition data, not charges records twice, but materializing all rows with list(rows) increases executor memory pressure

A common pattern for per-partition vault key sizing is to consume the iterator to compute the partition total, then iterate again to charge each record. But a Python iterator can only be consumed once — calling list(rows) to materialize all records for the sum, then iterating a second time, requires holding all partition records in executor memory simultaneously. For large partitions (millions of records), this can cause executor OOM, triggering exactly the task retry problem described in failure mode 1. Prefer computing billing totals in a separate Spark action (a groupBy aggregation) before the billing run, then broadcast the pre-computed totals for vault key sizing. This avoids double-materializing partition data inside the billing UDF and keeps executor memory usage bounded to one record at a time during the charge loop.

FAQ

Can we use the Spark task attempt ID as the Stripe idempotency key?

No. Spark's task attempt ID (accessible via TaskContext.get().attemptNumber()) increments on each retry: attempt 0, attempt 1, attempt 2. Using the attempt ID as or in the idempotency key means each retry sends a different key to Stripe, and Stripe creates a new charge on each retry — which is the exact failure mode you are trying to prevent. The idempotency key must be derived from the billing event's immutable business fields (customer ID, amount, billing period) so that every retry of every attempt sends the same key and receives the same charge object back. The task attempt ID can be useful for logging and debugging retry counts, but it must never be part of the idempotency key.

Does Databricks handle Spark task retries differently than open-source Spark?

Databricks uses the same underlying Spark task retry mechanism — executor failure causes task re-scheduling with the same spark.task.maxFailures default. Databricks does add automatic cluster restart and job retry policies at the cluster level (via the Jobs UI), which can trigger re-execution of entire job runs rather than individual tasks. A Databricks job configured with a retry policy of 2 attempts will re-run the entire billing job from scratch on failure, not just the failed tasks. Job-level retries require that your entire billing job is idempotent from start to finish: all Stripe calls must use content-hash idempotency keys, and the output sink must handle duplicate writes. The same governance pattern applies regardless of whether the retry comes from Spark task scheduling or Databricks job orchestration.

How does this apply to Delta Live Tables pipelines?

Delta Live Tables (DLT) pipelines in Databricks are triggered by the DLT runtime, which handles updates, retries, and incremental processing. DLT's @dlt.table and @dlt.view decorators define transformations that DLT may re-execute on pipeline updates or failures. If a DLT table definition includes a Python UDF that calls stripe.charges.create(), DLT may execute that UDF multiple times for the same records during pipeline refreshes or recovery. Content-hash idempotency keys apply identically to DLT pipelines: the key must be derived from the record's business fields, not from any DLT-managed metadata that changes between executions.

Does the two-pass approach (compute totals first, then issue vault keys, then charge) introduce a race condition if new customers are added between the two passes?

Yes, if the customer dataset is not snapshotted before the first pass. A new customer record added between the aggregation pass and the billing pass will appear in the billing partition but will not have been counted in the vault key sizing, potentially causing the vault key cap to be exceeded before all customers in the partition are charged. To eliminate this race condition, snapshot the input DataFrame before the aggregation: write it to a temporary path or use .cache() to pin the RDD in memory, then run the aggregation and billing passes against the same snapshot. Spark's lazy evaluation means that without an explicit snapshot, the two passes may read different data from the source if the source is mutable between passes.

What is the right vault key TTL for a Structured Streaming job that runs indefinitely?

For indefinitely-running streaming jobs, issue vault keys per micro-batch at the start of each foreachBatch() call rather than issuing a single key at job startup. Each call to issue_batch_vault_key() creates a fresh key scoped to that batch's expected total and capped with a short TTL (300 seconds is sufficient for most micro-batches). This approach ensures that each batch has a key sized to its actual records, the key expires naturally if the batch fails before completing, and a data error in one batch is contained to that batch's vault key cap without affecting subsequent batches. The overhead of one Keybrake API call per micro-batch is minimal relative to the Stripe API calls in the batch.

Can we use Spark's built-in foreachBatch exactly-once semantics to avoid needing idempotency keys?

Spark Structured Streaming does not provide exactly-once semantics for arbitrary external API calls in foreachBatch(). Spark's exactly-once guarantee applies only to supported sinks that implement idempotent writes (Delta Lake, Kafka with transactions, JDBC with upsert). For HTTP API calls — including Stripe — Spark makes no guarantee and explicitly documents that foreachBatch() may be called multiple times for the same batch on recovery. Idempotency keys at the Stripe API layer are the correct and only mechanism to prevent duplicate charges from checkpoint replay. There is no Spark configuration or sink option that provides HTTP-level exactly-once semantics.

Related: Ray Stripe Integration · Dagster Stripe Integration · Airflow Stripe Integration · Prefect Stripe Integration · AWS Step Functions Stripe Integration