Spark Kafka producer sending events in order

Question

A few days ago I posted a question where I wanted to know the optimal way to write events from Databricks to Kafka sorted by a timestamp field in ascending order.

Initially the objective was to develop it in the context of Structured Streaming. Finally, for business reasons, I had to do it in Batch. However, the objective does not change, the events have to be written in an orderly manner based on a timestamp field.

The input dataframe:

| field_id | ts    | 
|-----|------------|
| k2  | 2020-05-05 |
| k1  | 2020-05-04 |
| k3  | 2020-05-16 |  
| k1  | 2020-03-02 | 
| k2  | 2020-06-12 |
| k1  | 2020-06-01 |

Desired output dataframe:

| field_id| ts     | 
|-----|------------|
| k1  | 2020-03-02 |
| k1  | 2020-05-04 |
| k1  | 2020-06-01 |  
| k2  | 2020-05-05 | 
| k2  | 2020-06-12 |
| k3  | 2020-05-16 |

The code with which I intended to solve this issue was:

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._


val df = spark.read.format("parquet").load(globalPath)

val df_key = df.withColumn("key",struct($"field_id"))
val output_df = df_key.select($"key",struct($"field_id",$"ts").as("value")).orderBy($"key",$"ts".asc)


output_df
    .select(to_json($"key").as("key"),
            to_json($"value").as("value"))
    .write
    .format("kafka") 
    .option("kafka.bootstrap.servers", BOOSTRAP_SERVERS)
    .option("kafka.security.protocol", "SASL_SSL")
    .option("kafka.sasl.jaas.config", "kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";".format(username, password))
    .option("kafka.ssl.endpoint.identification.algorithm", "https")
    .option("kafka.sasl.mechanism", "PLAIN")
    .option("topic", TOPICNAME)
    .save()

The problem with this code is that I was not able to write in order in Kafka. To get around this, I added a repartition(1) before the .write and it worked. I got the order right. Indeed, this is not the best way to work with Spark so I decided to give it some more thought.

I came to two conclusions:

If instead of adding the orderBy when generating the output_df dataframe I put it just before the .write, it writes me in an orderly fashion. The output in Kafka is not exactly like in the desired output but for each key I get the ordered events (in Kafka shows in DESC. Latest TS first for each key). I think this is due to what is exposed here. It says ...If you call orderBy on a DataFrame and do not call any further operation which causes the DataFrame to shuffle , then the data in partitions stay sorted.... and ...after orderBy operation not only each partition is sorted from within but also among each other....
The other conclusion is that if I perform a repartition($"key"), Spark orders it in the same way as the previous point, so the output in Kafka is identical to the previous point.

I do not know if these conclusions imply any risk or if I am on the right track. I don't know the way Spark writes towards Kafka in detail. I do not quite understand how if there is a key present in two different partitions, it knows what is the order between one and another when I do the orderBy so that it is able to write it in order. Does it read from partition 0 and then go to partition 1 and then to partition 2 sequentially? or how does it do it?

In short, I do not know if this aproach can create me some problem or I am on the right track. Thanks

Spark Kafka producer sending events in order

0 Answers