topic Spark StreamingQuery not processing all data from source directory in Data Engineering

Spark StreamingQuery not processing all data from source directory

RajaLakshmanan — Fri, 20 Aug 2021 12:55:04 GMT

Hi,

I have setup a streaming process that consumers files from HDFS staging directory and writes into target location. Input directory continuesouly gets files from another process.

Lets say file producer produces 5 million records sends it to hdfs staging directory in 50 different files in 5 mintues winodw. The streaming process picks up the files ( processing trigger set to 60 secs) and processing them all but the problem is it is missing 2-3% records after processing all the files from input directory.

I checked the checkpoint source directory and it shows that all the files are picked up and processed.

Not sure what else to check.

@Prakash Chockalingam Can you pls help ?

Dataset<Row> dataDF = spark.readStream() .option("sep", delimiter) .option("header", "false") .option("inferSchema", "false") .option("latestFirst","false") .schema(appConfig.getSchema()) .csv(appConfig.getHdfsSourceDir());

StreamingQuery query = dataDF.writeStream() .trigger(ProcessingTime.create(60, TimeUnit.SECONDS)) .format("parquet") .outputMode(OutputMode.Append()) .partitionBy(appConfig.getPartitionedBy().split(",")) .option("checkpointLocation", appConfig.getHdfsNNUri() + appConfig.getHdfsOutputDir()+ "/checkpoint") .option("compression", appConfig.getSparkCompressType()) .start(appConfig.getHdfsNNUri() + appConfig.getHdfsOutputDir()); query.awaitTermination();

Re: Spark StreamingQuery not processing all data from source directory

jose_gonzalez — Tue, 28 Sep 2021 18:05:48 GMT

Hi,

Did you check the logs for the micro-batch metrics? these metrics will help you to identify the number of records you process and the time it took to process the data. This article shows the metrics that are capture and how you can use them to know how is your streaming job working. Link https://www.waitingforcode.com/apache-spark-structured-streaming/query-metrics-apache-spark-structured-streaming/read#measured_metrics

Thank you

Re: Spark StreamingQuery not processing all data from source directory

User16763506586 — Wed, 29 Sep 2021 11:12:37 GMT

If it helps , you run try running the Left-Anti join on source and sink to identify missing records and see whether the record is in match with the schema provided or not