spark技术分享KafkaWriteTask
KafkaWriteTask is used to write rows (from a structured query) to Apache Kafka.
KafkaWriteTask is created exclusively when KafkaWriter is requested to write the rows of a structured query to a Kafka topic.
KafkaWriteTask writes keys and values in their binary format (as JVM’s bytes) and so uses the raw-memory unsafe row format only (i.e. UnsafeRow). That is supposed to save time for reconstructing the rows to very tiny JVM objects (i.e. byte arrays).
| Name | Description |
|---|---|
|
|
|
|
|
|
|
|
Created once when |
Writing Rows to Kafka Asynchronously — execute Method
|
1 2 3 4 5 |
execute(iterator: Iterator[InternalRow]): Unit |
execute uses Apache Kafka’s Producer API to create a KafkaProducer and ProducerRecord for every row in iterator, and sends the rows to Kafka in batches asynchronously.
Internally, execute creates a KafkaProducer using Array[Byte] for the keys and values, and producerConfiguration for the producer’s configuration.
|
Note
|
execute creates a single KafkaProducer for all rows.
|
For every row in the iterator, execute uses the internal UnsafeProjection to project (aka convert) binary internal row format to a UnsafeRow object and take 0th, 1st and 2nd fields for a topic, key and value, respectively.
execute then creates a ProducerRecord and sends it to Kafka (using the KafkaProducer). execute registers a asynchronous Callback to monitor the writing.
|
Note
|
|
Creating UnsafeProjection — createProjection Internal Method
|
1 2 3 4 5 |
createProjection: UnsafeProjection |
createProjection creates a UnsafeProjection with topic, key and value expressions and the inputSchema.
createProjection makes sure that the following holds (and reports an IllegalStateException otherwise):
-
topicwas defined (either as the inputtopicor ininputSchema) and is of typeStringType -
Optional
keyis of typeStringTypeorBinaryTypeif defined -
valuewas defined (ininputSchema) and is of typeStringTypeorBinaryType
createProjection casts key and value expressions to BinaryType in UnsafeProjection.
|
Note
|
createProjection is used exclusively when KafkaWriteTask is created (as projection).
|
Creating KafkaWriteTask Instance
KafkaWriteTask takes the following when created:
KafkaWriteTask initializes the internal registries and counters.
