How Spark Structured Streaming reads and writes Kafka
Spark jobs that consume from or publish to a self-hosted Kafka cluster should use the Kafka source and sink provided by the spark-sql-kafka package, connected to the cluster's listeners as they are defined for your environment. That package is the integration path this topic covers, and it runs inside the same security boundary as the brokers, so the job needs network reachability to the bootstrap servers and the credentials your cluster requires. Spark does not hold a copy of the topic configuration or the ACLs. Those stay on the Kafka side, which means the team must confirm early which listener, protocol and SASL mechanism each job is allowed to use.
Spark tracks its own progress through checkpoints. For each query, the checkpoint location records the Kafka offsets that a micro-batch has read and processed, so a restarted job resumes from recorded offsets rather than from consumer group commits. Delivery guarantees are not a property of Spark alone. They depend on the sink and on how the job is designed: a Kafka sink can write records again after a failed batch is retried, so duplicates must be handled through idempotent keys, deduplication logic or a downstream store that tolerates replays. Teams should decide the guarantee they need for each pipeline before writing code, and they should ask whoever owns the platform to confirm how the checkpoint directory is stored and backed up in their infrastructure.
What you need before connecting Spark to Kafka
Connecting Spark Structured Streaming to Kafka touches three systems you usually do not own in the same team: the Spark runtime, the Kafka cluster and the network between them. The cards below cover the decisions and checks to settle with the platform team before the first query runs.
Choose the Spark release line
Spark 4.2.0 was released on July 14, 2026, and 4.1.3 and 4.0.4 followed as maintenance releases on July 15, 2026, with 3.5.9 on July 16, 2026. Spark 4 is pre-built with Scala 2.13 and has dropped Scala 2.12, so any application jars and the spark-sql-kafka package must match that Scala build. Confirm with the platform team which line your organization will support, and how long it expects to patch it.
Run a Kafka 4.x KRaft cluster
Kafka 4.0 and later run only in KRaft mode, so the cluster must not depend on ZooKeeper. Supported releases listed by the project are 4.3.1, 4.2.2 and 4.1.2. Ask the Kafka owners which version is in production and whether the bootstrap servers you will use are reachable from Spark.
Confirm listeners and TLS settings
Spark needs a broker listener that matches the security protocol your clients use, typically SASL_SSL with a SCRAM or OAUTHBEARER mechanism. Ask for the bootstrap address, the advertised listener names and the CA certificate chain, and confirm that broker hostnames in the certificates match what executors resolve. A plaintext listener may exist for internal tools, so do not assume it is the one Spark should use.
Check the Java runtime on every node
Brokers require Java 17, but Spark executors and the driver run on their own JVMs. Kafka clients and Kafka Streams applications need at least Java 11. Confirm the JVM version on each Spark worker, the driver and any container image, and make sure the Kafka client libraries bundled with your Spark distribution match the broker version you are targeting.
Map network paths from executors
Every Spark executor, not only the driver, opens its own connections to the brokers, so the firewall and routing rules must cover all worker nodes or pods. Kafka returns broker addresses from metadata, which means each advertised address must be reachable from every executor. Ask the network team to test from a worker node in each segment before you debug application code.
Decide on the identity source
Choose whether streaming jobs use a dedicated service account with SCRAM or Kerberos credentials, or short-lived delegation tokens issued to the Spark application. Delegation tokens suit long-running jobs that outlive a single credential, but they need a renewal process and a token owner. Ask the Kafka administrators which ACLs the service account needs for the topics, consumer groups and transactional IDs your jobs will use.
Configure the Spark Kafka source and sink in order
This sequence takes Spark from a dependency on the Kafka connector to a streaming query that reads a topic, tracks progress in a checkpoint and writes results to a sink. Each step ends with a check, so a fault is isolated before the next layer is added. Run it against a non-production cluster first.
Add the spark-sql-kafka package for your versions
The platform engineer adds the spark-sql-kafka package to the application build, choosing the artifact that matches the Spark release and the Scala binary version of the Spark installation. Spark 4 is pre-built with Scala 2.13 and has dropped Scala 2.12 support, so the Scala 2.13 artifact is the one to use on a Spark 4 cluster. Confirm the exact coordinates in the Spark Kafka integration documentation for your release, and deploy the same package to the driver and every executor. Done when the application starts and the Kafka data source resolves without a class-not-found error.
Create the test topic with planned partitions
The Kafka administrator creates a dedicated test topic using the partition count and replication factor from your capacity plan. Spark creates one read task per partition, so the partition count sets the maximum read parallelism for that topic. Ask the Kafka team to confirm the topic settings and to grant the Spark service identity read and write access through the broker authorization policy. Done when the topic describes with the planned partition count and the Spark identity can both produce and consume test records.
Produce a known set of test records
Load a fixed, countable set of records into the test topic, for example a few thousand JSON messages spread across several keys (illustrative). Record the expected total and the end offset of each partition before Spark reads anything, so the later counts have a reference. Use the producer tooling your team already runs. Done when the broker reports end offsets whose sum matches the number of records you sent.
Validate connectivity with a batch read
Run a batch read with spark.read.format('kafka') using the bootstrap servers, the topic name and the security settings for your listener, such as SASL_SSL with the mechanism your Kafka team has issued. Set startingOffsets to earliest and endingOffsets to latest so the read covers all test records. Count the rows and compare the result with the total from the previous step. A failure here points to network reachability, TLS trust or credentials, which is easier to diagnose than inside a running stream. Done when the row count matches the expected total exactly.
Start a streaming read with startingOffsets
Switch to spark.readStream with the same options and set startingOffsets deliberately. earliest replays the retained history, and latest starts from new records only. Use earliest for the test so every record is seen, and document the choice, because it decides what a first run without a checkpoint processes. Start the query with a console or memory sink and confirm that the batch progress reports the expected input rows. Done when the query shows input rows from the test topic and the Spark UI lists the Kafka source with its offset ranges.
Set the checkpoint location on durable storage
Set checkpointLocation to a path on storage that survives driver restarts, such as a shared file system or object store your team operates, and give each query its own directory. The checkpoint records source offsets and sink progress, so a restarted query resumes from the last committed batch. Ask the storage team to confirm durability and that the Spark driver can read and write the path. Done when the checkpoint directory contains offset and commit entries after the first micro-batches complete.
Write to the sink and confirm restart behavior
Write the stream to the target sink in a format or table your downstream team already reads, and attach the same checkpoint location. Add a second set of test records, stop the query and restart it from the checkpoint. Compare the sink contents with the two test totals. Any duplicate rows should be explained by the sink design, which the at-least-once and deduplication guidance covers. Done when the sink holds each test record once, and the checkpoint commit log advances after the restart.
How offsets and checkpoints decide what gets reprocessed
Spark's Kafka source records the offset it has processed for each topic partition inside the query checkpoint, so a restarted query resumes from the recorded position rather than from a fresh read. This makes the checkpoint the authoritative record of progress, and it is the first design decision to settle. The startingOffsets option (earliest, latest, or a JSON map of explicit offsets) only applies when a query starts with no checkpoint. Once a checkpoint exists, startingOffsets is ignored, so operators who expect a config change to rewind or skip data will be surprised. Treat a rewind as a deliberate operation: stop the query, decide whether to start a new checkpoint location, and document which offsets the new run will read.
Changing the checkpoint, the query or the topic can break resumption in ways that are not always obvious. Adding or removing stateful operators, changing aggregation keys, or altering the schema of stored state can make the saved checkpoint incompatible with the new plan, and Spark will fail at startup or require a new checkpoint. Changing the topic name or pointing at a recreated topic means recorded offsets may refer to data that no longer exists. Before any change, confirm the Spark version and query plan compatibility with the Spark Structured Streaming documentation for your release, and rehearse the change against a copy of the checkpoint in a non-production environment.
Offsets expire when Kafka's retention deletes the segments that hold them. If a query falls behind longer than retention, the recorded offset points to data that is gone. Spark's Kafka source can fail the query in this case, and the failOnDataLoss option controls whether that failure stops the query or only logs a warning and skips ahead. Setting it to false trades completeness for availability, so the choice belongs in the design review with the data owners, not in a default. The checkpoint directory must also live on storage that survives pod or node replacement, such as a persistent volume or an object store path that the platform supports for checkpoints. A checkpoint on ephemeral container storage means every reschedule becomes a silent reset to startingOffsets, which can mean reprocessing a full retention window or losing data. Confirm the checkpoint location, its backup and its access rights as part of the runbook.
Choosing write semantics: at-least-once and deduplication
When Spark writes to a Kafka topic through the spark-sql-kafka package, the default guarantee is at-least-once. Spark's micro-batch engine replays a batch from its checkpointed offsets after a failure, and the Kafka sink then sends the rows again. Whether those replays reach the topic as duplicates depends on the producer settings Spark uses, so the producer configuration is part of the write design, not an afterthought.
- Producer retries: Transient broker errors cause the producer to resend records. Ask whether idempotence is enabled for the sink's producer and confirm the behavior against the version of the spark-sql-kafka package you deploy.
- Batch replay: If a failure occurs after records are written but before the checkpoint commit, the batch is processed again and its rows are written again.
- Keys for downstream dedup: Set a stable key on each record so that consumers can collapse repeated values. Kafka keeps records with the same key in one partition, which helps ordering but does not remove duplicates.
- Consumer side handling: Consumers that need exactly-once results must deduplicate on a business key or use transactional reads, which the Kafka producer and consumer documentation describes.
- Retention and compaction: A compacted topic keeps the latest value per key, which can hide duplicates from later readers but is not a substitute for idempotent processing.
Before go-live, confirm the producer settings, the key scheme and the consumer-side rule for ignoring repeats with your platform team and the team that owns the downstream topic.
When Spark writes to a database or a lakehouse table, duplicates usually come from replayed micro-batches. A batch can be written to the target, fail before the checkpoint is committed, and then be processed again on restart. The sink does not know the batch was already applied, so the write design must make repeated writes harmless. The usual tools are a stable business key with an idempotent upsert, or a transactional table format that records the batch identity.
- Idempotent upserts: Write with a merge or upsert keyed on a unique business identifier so that a replayed row overwrites itself instead of adding a new one.
- Deterministic keys: Derive keys from the source record (for example a Kafka topic, partition and offset, or a domain identifier), not from a random value generated during processing.
- Watermarks for state: Use
withWatermarkto bound the state kept for deduplication withdropDuplicateson streaming data, so state does not grow without limit. Late records outside the watermark are dropped by that logic. - Batch identity: Some lakehouse formats let a writer record the batch or query identity so that an already committed batch is skipped. Confirm the exact behavior for your table format and engine version.
- Database constraints: Unique constraints catch duplicates that the merge logic misses, and a failed insert then surfaces as an error to investigate.
Exactly-once results across the Kafka source and this sink depend on the sink's commit behavior, so test a forced failure in a non-production environment before relying on the design.
The foreachBatch method gives the application a function that receives each micro-batch as a static DataFrame along with a batch identifier. It is the usual choice when the target is a system that Spark has no native streaming sink for, or when one batch must be written to several targets. The guarantees are set by the code you write inside the function, so duplicate handling is entirely the application's responsibility.
- Batch identifier: Use the batch ID that Spark passes to the function to tag writes, and record the last applied ID in the target so that a replayed batch can be detected and skipped.
- Non-atomic multi-target writes: Writing to two systems in one function can apply one and fail on the other, so replays may double-apply the first write.
- Idempotent target operations: Prefer upserts or overwrite of a partition for a batch ID over plain inserts, since a retried function call then produces the same result.
- Transactional sinks: Where the target supports transactions, commit the data and the batch ID together so that both succeed or both roll back.
- Side effects: Calls to external APIs, emails or message queues repeat on replay unless each call is keyed by batch and record, so keep them out of the function or make them idempotent.
Exactly-once behavior with external systems is not provided by Spark alone. Document which target operations are idempotent, and have the team that owns the target confirm how it handles repeated batch IDs.
Securing the connection with SASL_SSL and delegation tokens
A Spark job talks to Kafka through the same security layers as any other client: TLS for encryption, a SASL mechanism for identity, and ACLs for what that identity may touch. Delegation tokens matter because streaming jobs run far longer than a typical client session, so the cards below cover each layer and the checks to run with the cluster owner.
TLS encryption in transit
Kafka encrypts traffic between clients, brokers and tools with SSL/TLS, and security is optional, so a cluster can run unsecured or mix secured and unsecured listeners. For Spark, the Kafka source and sink must connect to a SASL_SSL listener, not a plaintext one. Ask the cluster owner for the bootstrap server port that carries SASL_SSL, the CA bundle the Spark executors must trust, and whether the listener requires client certificates. Check that the truststore is distributed to every driver and executor, not only the driver.
Choosing a SASL mechanism
Kafka documents GSSAPI (Kerberos), PLAIN, SCRAM-SHA-256, SCRAM-SHA-512 and OAUTHBEARER for SASL authentication. Spark passes the mechanism through the Kafka client options, so the choice must match what the brokers enable. SCRAM credentials are stored in the cluster and suit service accounts, while OAUTHBEARER fits organizations that already issue tokens from an identity provider. Confirm the enabled mechanisms and the listener name with the cluster owner before writing any job configuration.
Delegation tokens for long-running jobs
A structured streaming query can run for months, and a password or Kerberos ticket may expire while it is still active. Delegation tokens let a job authenticate to Kafka without carrying the main credential into every executor. Confirm with the cluster owner that delegation tokens are enabled, who can create them, their maximum lifetime and how often they must be renewed. Plan the renewal process and test a query that runs past the first token expiry before going to production.
Token renewal and job restarts
A token that cannot renew will stop a query, and the failure may appear only after days of normal operation. Decide who owns renewal, whether the Spark application or the cluster scheduler handles it, and what happens when renewal fails. Test a forced token expiry in a non-production environment and confirm that the query recovers from its checkpoint with the correct offsets. Alert on authentication errors in the driver logs so that a stopped query is noticed quickly.
ACLs for the Spark principal
The Spark principal needs read access to source topics and the consumer group it uses, and write access to sink topics. Each Spark query uses its own consumer group, so ACLs must cover those group names as well as the topics. Resources without ACLs are restricted to super users in KRaft clusters unless allow.everyone.if.no.acl.found is set to true. Ask the cluster owner to confirm the authorizer settings and to grant topic and group rights with the narrowest scope the job needs.
Authorizer settings to confirm
In KRaft clusters, the StandardAuthorizer stores ACLs in the cluster metadata log. Before testing, ask the cluster owner which authorizer is active, whether allow.everyone.if.no.acl.found is set, and whether an external authorization service is in use. A Spark job that works in a test cluster with permissive defaults can fail in production once those settings are tightened. Record the answers with the job configuration so the next change to the cluster does not break the query without warning.
Mapping Spark options to Kafka settings and identities
This table maps each Spark Kafka source and sink option to the client setting, principal and ACL it produces. Read it when you define the identity a Spark job runs under, since the same option can require different permissions on the read and write sides. Confirm the exact behavior for your Spark and Kafka versions in the package documentation before you write the ACLs.
| Spark option | Kafka effect | ACL and identity consideration |
|---|---|---|
| subscribe or subscribePattern | Selects topics explicitly or by regex; the source assigns partitions and tracks offsets for them | Read on each topic; with subscribePattern, confirm the principal can describe every matching topic, since new matches appear over time |
| startingOffsets | Sets where a new query begins (earliest, latest or per-partition JSON); it applies only when no checkpoint exists | Changing it after a checkpoint exists has no effect; plan replays by new checkpoint location, not by editing this option |
| kafka.bootstrap.servers | Sets the initial broker list the client uses to discover the cluster | Network path and listener must match the security protocol; a plaintext listener cannot serve a SASL_SSL client |
| kafka.security.protocol | Sets SASL_SSL, SASL_PLAINTEXT, SSL or PLAINTEXT on the client connection | Must match a listener the broker actually exposes; mismatches typically surface as connection or authentication failures |
| kafka.sasl.mechanism | Selects the SASL mechanism, such as SCRAM-SHA-512, GSSAPI or OAUTHBEARER, from those the cluster enables | The mechanism determines which principal the broker sees; confirm it with the platform team before wiring credentials |
| Consumer group (group.id) | Spark sets a generated group per query unless configured; offsets are tracked through checkpoints, not committed group offsets | Grant Read on the group resource where the principal needs it; confirm how your Spark version names and isolates groups before writing ACLs |
| checkpointLocation | Stores offsets and query progress for recovery; it is not a Kafka setting | Must be durable storage the driver can write; a lost or shared checkpoint directory changes which offsets a restart resumes from |
| Sink topic (kafka topic option) | Writes records to the named topic; the sink does not take a consumer group | Requires Write and Describe on the topic; grant Create only if you intend the job to create topics |
| Transactional identity | Spark's Kafka sink does not expose a transactional.id option; confirm current support before planning exactly-once writes | Do not assume transactional fencing across restarts; confirm the delivery semantics your Spark version provides |
Questions teams ask before connecting Spark to Kafka
What drives the effort and cost of a Spark and Kafka integration?
The biggest drivers are the number of topics and sinks, the message format and schema handling, the amount of stateful logic, and how many security layers the integration must pass through. A job that reads one topic with a simple JSON parse and writes to one table is far smaller than a pipeline with joins, watermarks, deduplication and several sinks. Cost has two parts: one-time design, build and test work, and recurring operations such as upgrades, checkpoint handling and on-call. Ask for a quote that lists each source, sink, offset and checkpoint strategy, the authentication method (for example SASL_SSL with delegation tokens), and what the client's team will own after handover.
How long does a Spark and Kafka integration take?
No duration is given here, because the answer depends on the sources, sinks, security model and how much of the pipeline logic is new. Duration rises with each additional topic, each external sink with its own exactly-once or upsert behavior, and each access control layer that must be reviewed. Ask for a written plan with phases, the environments included (development, test and production), the acceptance tests for offsets and recovery, and the client decisions that must be made before build starts. Any timeline should be tied to a fixed scope agreed in writing.
What does the client's team need to provide?
The work needs access and decisions more than code. Expect to provide the Kafka cluster details and listener endpoints, the topic names and expected message schemas, a named owner for each topic and sink, a Spark cluster or runtime where jobs will run, and a way to store and rotate credentials. Your security team should confirm the authentication mechanism, the TLS trust chain and the ACLs that allow the Spark principal to read and write. Your data team should confirm dedup keys, retention expectations and how late data should be handled.
Can an internal team build this without a partner?
Yes, if the team already runs Spark in production and understands Kafka offsets, consumer groups and checkpoints. The skills that most often decide success are restarting a streaming query from its checkpoint, choosing an at-least-once or idempotent write pattern, and setting up SASL_SSL and delegation tokens correctly in a cluster that the team operates. A partner is most useful when the team is new to one of these areas, when a fixed delivery date matters, or when the team needs an independent review of the design before it goes live. Whichever path you choose, the design decisions should be written down and owned by someone inside your organization.
Which Spark and Kafka versions should we pick?
Pick a Kafka release from the supported list on the Kafka downloads page, and a Spark line that is still maintained, then confirm that the spark-sql-kafka package matches your Spark Scala version. Spark 4 is built with Scala 2.13 and has dropped Scala 2.12, so a Scala 2.12 Spark job will need a migration first. Spark's versioning policy says major and feature branches are maintained for 6 months, and the last release in each major line is an LTS maintained for 18 months. Check the Spark downloads page for current release dates before you choose. Also check that your Java runtime meets the requirements in the Kafka upgrade notes, and test your chosen pair in a non-production environment before going live.
How does delegation token renewal affect long-running Spark jobs?
Delegation tokens expire, so a streaming query that runs for weeks can fail when its token is not renewed. Spark can obtain and renew Kafka delegation tokens for the application, but the renewal interval, the token lifetime set on the Kafka broker, and the principal used must all be configured to match. Confirm with your Kafka administrator how token maximum lifetime and renewal are set, and confirm that the Spark driver can reach the cluster when renewal runs. Test a renewal cycle before production, including a restart from checkpoint, and set alerts on authentication failures so a stalled job is noticed quickly. For the exact configuration keys, check the Spark Kafka integration documentation for your chosen version.
What goes wrong most often with offsets and checkpoints?
The most common failure is changing a query and restarting it from an old checkpoint, which can fail or replay data unexpectedly. Other frequent problems include deleting the checkpoint directory to clear an error, which loses the record of processed offsets, and writing to an external sink without a dedup key, which creates duplicates on retry. Keep checkpoints on durable storage that all restarts can reach, record the starting offsets policy for each new query, and document how to reprocess a range safely. Ask for a restart test as part of acceptance, including a planned upgrade of the Spark version.
How to test the pipeline before production traffic
Test offset recovery before any production traffic arrives, because a Structured Streaming query that restarts from the wrong place either drops records or reprocesses them. Run the query against a test topic, let it commit several micro-batches, stop it, and restart it from the same checkpointLocation. Confirm that the first batch after restart begins at the offset following the last committed one, and that the sink contains each input record exactly once after the sink's own deduplication logic is applied. Then repeat the test after changing the query's startingOffsets value on purpose, to confirm the team understands which setting wins when a checkpoint already exists. Keep a written record of the expected behavior so the on-call engineer has a reference during an incident.
Next, inject failures on the parts of the path that fail in production. Stop a broker or move partition leadership during a run and watch the query's retry behavior and its driver and executor logs. Make the sink unavailable for a short period, for example by blocking its network route or pausing the database, and confirm the query fails cleanly, restarts, and does not advance its checkpoint past uncommitted output. For at-least-once writes, replay a known input window into the source and check that the deduplication key you chose, applied in the sink or in a dropDuplicatesWithWatermark step, produces one output row per business key. Document which failures the query survives without intervention and which need a restart, since that boundary sets your alerting.
Finish with security and capacity checks. Run the query with kafka.security.protocol set to SASL_SSL and confirm that a client without the right credentials, or with a principal that lacks read or write ACLs on the topic and consumer group, is refused with a clear error rather than a silent stall. Check delegation token renewal if you use tokens, including a run that spans the renewal window. For load, run against a topic with the partition count you plan to use in production and measure whether maxOffsetsPerTrigger and the executor count keep batch duration below your trigger interval. Confirm these figures with your own cluster and workload, since partition sizing depends on your data volume and Spark's own resource settings.
How DeployOpen scopes and delivers Spark and Kafka integrations
DeployOpen plans, implements, hardens, documents and supports self-hosted open-source platforms inside your own cloud accounts, data centers or air-gapped networks. For Spark Structured Streaming with Kafka, we start from a written scope covering the Kafka cluster, the Spark runtime, the checkpoint storage and the security settings that connect them.
How we scope
- Engagements are fixed-scope, with scope, deliverables and timelines agreed in writing before work starts.
- The written scope lists the Spark version, Kafka cluster, topics, checkpoint storage and sink systems involved.
- We confirm which SASL mechanism, delegation token setup and TLS settings your Kafka brokers already expose.
How we build
- We build and test streaming jobs against your Kafka topics, covering offset handling, checkpoint recovery and restart behavior.
- Write paths are tested for at-least-once delivery and deduplication logic agreed with your data owners.
- Security testing covers the SASL_SSL configuration and the credentials each job uses to reach Kafka.
How we hand over
- Deliverables include configuration, job definitions, runbooks for checkpoint recovery and a record of the decisions made.
- We document how to resume a job from a checkpoint, reset offsets and rotate delegation tokens in your environment.
- Handover sessions walk your platform and data engineering teams through operating the jobs without DeployOpen.
How we support
- Ongoing support, upgrades and continuous testing are available as retainers after handover.
- Retainers can cover Spark and Kafka version upgrades, checked against the release notes for each version you run.
- Ask any provider to confirm how they test restarts, failover and the upgrade path before agreeing the scope.
