spark structured streaming operation duration

Viewed 377

I am running a structured streaming job with kafka source.
spark: 2.4.7
python: 3.6.8

spark = SparkSession.builder.getOrCreate()

ds = (spark.readStream
           .format("kafka")
           .option("kafka.bootstrap.servers", kafka_servers)
           .option("subscribe", topic_name)
           .load())

# data preprocessing
ds = ...

model = GBTClassificationModel.load(model_path)

ds = model.transform(ds)  

query = (ds.writeStream
           .outputMode("update")
           .format("kafka")
           .option("kafka.bootstrap.servers", kafka_servers)
           .option("checkpointLocation", checkpoint_dir)
           .option("topic", output_topic)
           .trigger(processingTime="0 seconds")
           .start())

query.awaitTermination()

The spark web UI displays the following indicators:

AddBatch: 1955.0
GetBatch: 1.0
GetOffset: 0.0
QueryPlanning: 3555.0
TriggerExecution: 2.0
WalCommit: 5569.0
undefined: 20.0

The following is a description of the indicators on the spark official website:

Operation Duration. The amount of time taken to perform various operations in milliseconds. The tracked operations are listed as follows.

  1. addBatch: Time taken to read the micro-batch’s input data from the sources, process it, and write the batch’s output to the sink. This should take the bulk of the micro-batch’s time.
  2. getBatch: Time taken to prepare the logical query to read the input of the current micro-batch from the sources.
  3. latestOffset & getOffset: Time taken to query the maximum available offset for this source.
  4. queryPlanning: Time taken to generates the execution plan.
  5. walCommit: Time taken to write the offsets to the metadata log.

Why are WalCommit and QueryPlanning much larger than AddBatch?

Thanks!

0 Answers
Related