I have written a simple kafka pyspark structured streaming., by reading the kafka topic and print each event's partition and offset id using pyspark foreach function. But program is failing after reading few offsets.
class RowPrinter:
def open(self, partition_id, epoch_id):
print("Opened %d, %d" % (partition_id, epoch_id))
return True
def process(self, row):
print("partition: ", row[4], "offset:", row[5])
def close(self, error):
print("Closed with error: %s" % str(error))
spark = SparkSession.builder.appName("Structured Streaming ").master("local[*]").getOrCreate()
df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "XX.XX.XX.XX:9092,XX.XX.XX.XX.XX:9092").option("subscribe", "log_pst").option("failOnDataLoss", False).load()
test_df = df.selectExpr("topic", "key", "value", "timestamp", "partition", "offset")
query = test_df .writeStream.foreach(RowPrinter()).start()
query.awaitTermination()
Running the file using the command:
spark-submit --jars "D:\kafkaJars\spark-sql-kafka-0-10_2.12-3.0.0.jar,D:\kafkaJars\kafka-clients-3.0.0.jar,D:\kafkaJars\spark-token-provider-kafka-0-10_2.12-3.0.0.jar,D:\kafkaJars\commons-pool2-2.11.1.jar" --name example-streaming D:\stream.py
Getting below o/p and error:
Opened 2, 1
partition: 8 offset: 7304836
partition: 8 offset: 7304837
partition: 8 offset: 7304838
Opened 5, 1
partition: 2 offset: 7314983
partition: 2 offset: 7314984
partition: 2 offset: 7314985
partition: 2 offset: 7314986
partition: 2 offset: 7314987
partition: 2 offset: 7314988
partition: 2 offset: 7314989
Opened 8, 1
partition: 0 offset: 7296550
partition: 0 offset: 7296551
partition: 0 offset: 7296552
partition: 2 offset: 7314990
partition: 2 offset: 7314991
partition: 2 offset: 7314992
partition: 2 offset: 7314993
partition: 2 offset: 7314994
partition: 2 offset: 7314995
partition: 2 offset: 7314996
partition: 2 offset: 7314997
Opened 9, 1
partition: 3 offset: 7299538
Opened 3, 1
partition: 5 offset: 7308446
Closed with error: timed out
22/02/14 10:36:23 ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 7)
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "C:\hadoop\python\lib\pyspark.zip\pyspark\worker.py", line 605, in main
File "C:\hadoop\python\lib\pyspark.zip\pyspark\worker.py", line 595, in process
File "C:\hadoop\python\lib\pyspark.zip\pyspark\sql\streaming.py", line 1113, in func_with_open_process_close
File "C:\hadoop\python\lib\pyspark.zip\pyspark\sql\streaming.py", line 1105, in func_with_open_process_close
File "C:\hadoop\python\lib\pyspark.zip\pyspark\serializers.py", line 147, in load_stream
yield self._read_with_length(stream)
File "C:\hadoop\python\lib\pyspark.zip\pyspark\serializers.py", line 164, in _read_with_length
length = read_int(stream)
File "C:\hadoop\python\lib\pyspark.zip\pyspark\serializers.py", line 593, in read_int
length = stream.read(4)
File "C:\Users\xxx\AppData\Local\Programs\Python\Python37\lib\socket.py", line 589, in readinto
return self._sock.recv_into(b)
socket.timeout: timed out
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:503)
at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:638)
at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:621)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:456)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at org.apache.spark.sql.execution.python.PythonForeachWriter.close(PythonForeachWriter.scala:66)
at org.apache.spark.sql.execution.streaming.sources.ForeachDataWriter.close(ForeachWriterTable.scala:160)
at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$.$anonfun$run$15(WriteToDataSourceV2Exec.scala:477)
at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$$$Lambda$2711/1831861092.apply$mcV$sp(Unknown Source)
at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1433)
at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$.run(WriteToDataSourceV2Exec.scala:477)
at org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.$anonfun$writeWithV2$2(WriteToDataSourceV2Exec.scala:385)
at org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec$$Lambda$2433/262393842.apply(Unknown Source)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:127)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:463)
at org.apache.spark.executor.Executor$TaskRunner$$Lambda$2295/1718592933.apply(Unknown Source)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:466)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
22/02/14 10:36:23 ERROR TaskSetManager: Task 0 in stage 1.0 failed 1 times; aborting job
22/02/14 10:36:23 ERROR WriteToDataSourceV2Exec: Data source write support org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@69ff7916 is aborting.
22/02/14 10:36:23 ERROR WriteToDataSourceV2Exec: Data source write support org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@69ff7916 aborted.
22/02/14 10:36:23 ERROR MicroBatchExecution: Query [id = 610a3553-1eed-4136-8e45-bfe05bd8a079, runId = cb02e06f-abef-4c86-a15a-1aa803055cbb] terminated with error
org.apache.spark.SparkException: Writing job aborted.
at org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.writeWithV2(WriteToDataSourceV2Exec.scala:413)
at org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.writeWithV2$(WriteToDataSourceV2Exec.scala:361)
at org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec.writeWithV2(WriteToDataSourceV2Exec.scala:322)
at org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec.run(WriteToDataSourceV2Exec.scala:329)
at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result$lzycompute(V2CommandExec.scala:39)
at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result(V2CommandExec.scala:39)
at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.executeCollect(V2CommandExec.scala:45)
at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3627)
at org.apache.spark.sql.Dataset.$anonfun$collect$1(Dataset.scala:2940)
at org.apache.spark.sql.Dataset$$Lambda$1805/1235661287.apply(Unknown Source)
at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3618)
at org.apache.spark.sql.Dataset$$Lambda$1806/2028141464.apply(Unknown Source)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100)
at org.apache.spark.sql.execution.SQLExecution$$$Lambda$1791/1316458534.apply(Unknown Source)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87)
at org.apache.spark.sql.execution.SQLExecution$$$Lambda$1784/2095663039.apply(Unknown Source)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:767)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3616)
at org.apache.spark.sql.Dataset.collect(Dataset.scala:2940)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runBatch$16(MicroBatchExecution.scala:586)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$Lambda$1783/1737764274.apply(Unknown Source)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100)
at org.apache.spark.sql.execution.SQLExecution$$$Lambda$1791/1316458534.apply(Unknown Source)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87)
at org.apache.spark.sql.execution.SQLExecution$$$Lambda$1784/2095663039.apply(Unknown Source)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:767)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runBatch$15(MicroBatchExecution.scala:581)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$Lambda$1782/1042917224.apply(Unknown Source)
at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:352)
at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:350)
at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:69)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runBatch(MicroBatchExecution.scala:581)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:223)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$Lambda$1376/397443005.apply$mcV$sp(Unknown Source)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:352)
at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:350)
at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:69)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:191)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$Lambda$1374/52182715.apply$mcZ$sp(Unknown Source)
at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:57)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:185)
at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:334)
at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:245)
Can someone tell me why its failing and what configuration have I missed.