Kafka + Spark

The Bulletproof Kafka → Spark Query

kafka_df = (spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka-prod:9092")
    .option("subscribe", "user_events,orders")
    .option("startingOffsets", "earliest")          # dev
    # .option("startingOffsets", "latest")          # prod (or use checkpoint)
    .option("failOnDataLoss", "false")              # survive topic deletion
    .option("kafka.security.protocol", "SASL_SSL")  # if needed
    .option("maxOffsetsPerTrigger", "100000")       # rate limiting / backpressure
    .load()
)

parsed = (kafka_df
    .select(
        col("key").cast("string"),
        from_json(col("value").cast("string"), schema).alias("json")
    )
    .select("key", "json.*", "topic", "partition", "offset", "timestamp")
    .withColumn("processing_time", current_timestamp())
)

query = (parsed
    .writeStream
    .format("delta")
    .option("checkpointLocation", "s3a://checkpoints/kafka-events/")
    .partitionBy("date", "topic")
    .trigger(processingTime="60 seconds")
    .outputMode("append")
    .start("s3a://delta/events_bronze")
)

query.awaitTermination()
            

Exactly-Once = Two Things Only

Options That Save Your Job

# Production must-haves
.option("failOnDataLoss", "false")                    # survive topic recreation
.option("maxOffsetsPerTrigger", "50000")              # backpressure
.option("minPartitions", "100")                       # parallelism
.option("includeHeaders", "true")                     # if you need headers

# Security (AWS/MSK/Confluent)
.option("kafka.security.protocol", "SASL_SSL")
.option("kafka.sasl.mechanism", "AWS_MSK_IAM")
.option("kafka.sasl.jaas.config", "software.amazon.msk.auth.MSKIAMAwareClientCallbackHandler")
.option("kafka.sasl.client.callback.handler.class", "software.amazon.msk.auth.iam.IAMClientCallbackHandler")
            

Backpressure That Actually Works

Production Checklist (Never Lose Data)