0
# Apache Spark Kafka Integration
1
2
Apache Spark Kafka Integration provides comprehensive structured streaming and batch data processing capabilities for Apache Kafka. This module enables seamless reading from and writing to Kafka topics using Spark DataFrames and Datasets with support for micro-batch processing, continuous streaming, and batch operations with complete offset management and fault tolerance.
3
4
## Package Information
5
6
- **Package Name**: spark-sql-kafka-0-10_2.11
7
- **Package Type**: maven
8
- **Language**: Scala
9
- **Installation**: Add to your Spark application dependencies
10
- **Coordinate**: `org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.8`
11
12
## Core Imports
13
14
```scala
15
import org.apache.spark.sql.SparkSession
16
import org.apache.spark.sql.streaming.Trigger
17
import org.apache.kafka.common.TopicPartition
18
```
19
20
## Basic Usage
21
22
### Reading from Kafka (Streaming)
23
24
```scala
25
import org.apache.spark.sql.SparkSession
26
27
val spark = SparkSession.builder
28
.appName("KafkaExample")
29
.getOrCreate()
30
31
// Read from Kafka topic
32
val df = spark
33
.readStream
34
.format("kafka")
35
.option("kafka.bootstrap.servers", "localhost:9092")
36
.option("subscribe", "my-topic")
37
.option("startingOffsets", "latest")
38
.load()
39
40
// Process the stream
41
val query = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
42
.writeStream
43
.outputMode("append")
44
.format("console")
45
.trigger(Trigger.ProcessingTime("10 seconds"))
46
.start()
47
48
query.awaitTermination()
49
```
50
51
### Writing to Kafka
52
53
```scala
54
// Write DataFrame to Kafka
55
df.select(
56
col("id").cast("string").as("key"),
57
to_json(struct(col("*"))).as("value")
58
)
59
.write
60
.format("kafka")
61
.option("kafka.bootstrap.servers", "localhost:9092")
62
.option("topic", "output-topic")
63
.save()
64
```
65
66
### Batch Processing
67
68
```scala
69
// Read from Kafka for batch processing
70
val batchDF = spark
71
.read
72
.format("kafka")
73
.option("kafka.bootstrap.servers", "localhost:9092")
74
.option("subscribe", "my-topic")
75
.option("startingOffsets", "earliest")
76
.option("endingOffsets", "latest")
77
.load()
78
```
79
80
## Architecture
81
82
The Spark Kafka integration is built around several key components:
83
84
- **Data Source Provider**: `KafkaSourceProvider` implements Spark's DataSource API for both V1 and V2
85
- **Consumer Strategies**: Flexible patterns for topic assignment (Assign, Subscribe, SubscribePattern)
86
- **Offset Management**: Comprehensive offset tracking with configurable start/end positions
87
- **Streaming Readers**: Micro-batch and continuous processing capabilities
88
- **Producer Integration**: Efficient writing with connection pooling and caching
89
- **Schema Management**: Fixed schema for Kafka records with proper type handling
90
91
## Kafka Record Schema
92
93
All Kafka records follow this fixed schema:
94
95
```scala
96
StructType(Seq(
97
StructField("key", BinaryType), // Message key (nullable)
98
StructField("value", BinaryType), // Message value (nullable)
99
StructField("topic", StringType), // Topic name
100
StructField("partition", IntegerType), // Partition number
101
StructField("offset", LongType), // Message offset
102
StructField("timestamp", TimestampType), // Message timestamp
103
StructField("timestampType", IntegerType) // Timestamp type (0=CreateTime, 1=LogAppendTime)
104
))
105
```
106
107
## Capabilities
108
109
### Consumer Strategies
110
111
Flexible patterns for consuming data from Kafka topics, supporting subscription by topic names, regex patterns, or specific partition assignments.
112
113
```scala { .api }
114
sealed trait ConsumerStrategy {
115
def createConsumer(kafkaParams: ju.Map[String, Object]): Consumer[Array[Byte], Array[Byte]]
116
}
117
118
case class AssignStrategy(partitions: Array[TopicPartition]) extends ConsumerStrategy
119
case class SubscribeStrategy(topics: Seq[String]) extends ConsumerStrategy
120
case class SubscribePatternStrategy(topicPattern: String) extends ConsumerStrategy
121
```
122
123
[Consumer Strategies](./consumer-strategies.md)
124
125
### Offset Management
126
127
Comprehensive offset tracking and range limit handling for precise control over data consumption boundaries.
128
129
```scala { .api }
130
sealed trait KafkaOffsetRangeLimit
131
case object EarliestOffsetRangeLimit extends KafkaOffsetRangeLimit
132
case object LatestOffsetRangeLimit extends KafkaOffsetRangeLimit
133
case class SpecificOffsetRangeLimit(partitionOffsets: Map[TopicPartition, Long]) extends KafkaOffsetRangeLimit
134
135
case class KafkaSourceOffset(partitionToOffsets: Map[TopicPartition, Long]) extends OffsetV2
136
```
137
138
[Offset Management](./offset-management.md)
139
140
### Streaming Data Sources
141
142
Advanced streaming readers supporting both micro-batch and continuous processing modes with fault tolerance and exactly-once semantics.
143
144
```scala { .api }
145
class KafkaMicroBatchReader extends MicroBatchReader with Logging {
146
def setOffsetRange(start: Option[Offset], end: Offset): Unit
147
def planInputPartitions(): ju.List[InputPartition[InternalRow]]
148
def readSchema(): StructType
149
}
150
151
class KafkaContinuousReader extends ContinuousReader with Logging {
152
def readSchema: StructType
153
def setStartOffset(start: Option[Offset]): Unit
154
def planInputPartitions(): ju.List[InputPartition[InternalRow]]
155
}
156
```
157
158
[Streaming Sources](./streaming-sources.md)
159
160
### Batch Data Access
161
162
Batch relation for reading historical data from Kafka topics with configurable offset ranges.
163
164
```scala { .api }
165
class KafkaRelation extends BaseRelation with TableScan with Logging {
166
def sqlContext: SQLContext
167
def schema: StructType
168
def buildScan(): RDD[Row]
169
}
170
```
171
172
[Batch Processing](./batch-processing.md)
173
174
### Data Writing
175
176
Comprehensive writing capabilities for both streaming and batch workloads with producer connection pooling and automatic serialization.
177
178
```scala { .api }
179
class KafkaSink extends Sink with Logging {
180
def addBatch(batchId: Long, data: DataFrame): Unit
181
}
182
183
class KafkaStreamWriter extends StreamWriter {
184
def createWriterFactory(): KafkaStreamWriterFactory
185
def commit(epochId: Long, messages: Array[WriterCommitMessage]): Unit
186
}
187
188
object KafkaWriter extends Logging {
189
def write(sparkSession: SparkSession, queryExecution: QueryExecution,
190
kafkaParameters: ju.Map[String, Object], topic: Option[String]): Unit
191
}
192
```
193
194
[Data Writing](./data-writing.md)
195
196
### Configuration and Options
197
198
Complete configuration options for fine-tuning Kafka integration behavior, connection parameters, and performance settings.
199
200
**Source Options**:
201
```scala { .api }
202
// Connection
203
"kafka.bootstrap.servers" -> "localhost:9092"
204
"subscribe" -> "topic1,topic2"
205
"subscribePattern" -> "topic.*"
206
"assign" -> """{"topic1":[0,1],"topic2":[0]}"""
207
208
// Offset Management
209
"startingOffsets" -> "earliest" // or "latest" or JSON
210
"endingOffsets" -> "latest" // or JSON (batch only)
211
"failOnDataLoss" -> "true"
212
213
// Performance
214
"minPartitions" -> "10"
215
"maxOffsetsPerTrigger" -> "1000000"
216
```
217
218
**Sink Options**:
219
```scala { .api }
220
"kafka.bootstrap.servers" -> "localhost:9092"
221
"topic" -> "output-topic"
222
```
223
224
[Configuration](./configuration.md)
225
226
## Types
227
228
```scala { .api }
229
// Package-level type alias
230
type PartitionOffsetMap = Map[TopicPartition, Long]
231
232
// Data Consumer Types
233
case class AvailableOffsetRange(earliest: Long, latest: Long)
234
235
sealed trait KafkaDataConsumer {
236
def get(offset: Long, untilOffset: Long, pollTimeoutMs: Long,
237
failOnDataLoss: Boolean): ConsumerRecord[Array[Byte], Array[Byte]]
238
def getAvailableOffsetRange(): AvailableOffsetRange
239
def release(): Unit
240
}
241
242
// Offset Range Types
243
case class KafkaOffsetRange(
244
topicPartition: TopicPartition,
245
fromOffset: Long,
246
untilOffset: Long,
247
preferredLoc: Option[String]
248
) {
249
lazy val size: Long = untilOffset - fromOffset
250
}
251
252
// RDD Types
253
case class KafkaSourceRDDOffsetRange(
254
topicPartition: TopicPartition,
255
fromOffset: Long,
256
untilOffset: Long,
257
preferredLoc: Option[String]
258
) {
259
def topic: String = topicPartition.topic
260
def partition: Int = topicPartition.partition
261
def size: Long = untilOffset - fromOffset
262
}
263
```