Hi all:
I am having a strange issue incorporating `groupBy` statements into a
structured streaming job when trying to write to Kafka or Delta. Weirdly it
only appears to work if I write to console, or to memory...
*I'm running Spark 3.0.1 with the following dependencies:
*
io.delta:delta-core_2.12:0.7.0
org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.1
org.apache.hadoop:hadoop-azure:3.2.1"
*Here's a example of the pyspark job I've been testing with:*
/kafka = spark.readStream\
.format("kafka")\
.option("kafka.bootstrap.servers","kafka-broker:29092")\
.option("subscribe","test")\
.option("startingOffsets", "earliest")\
.option("maxOffsetsPerTrigger", 5) \
.load()
rawDF = kafka.selectExpr("CAST(value AS STRING)")
groupDF = rawDF.groupBy("value").agg(count(lit(1)).alias("count"))
kafka_stream_output = groupDF.selectExpr("to_json(struct(*)) AS value")) AS
value")
kafka_stream_output \
.writeStream \
.format("kafka") \
.outputMode("update") \
.option("kafka.bootstrap.servers", "kafka-broker:29092") \
.option("topic", "sink") \
.option("checkpointLocation", checkpoint_location) \
.start()/
*
If I don't have a groupBy/aggregation, it's able to stream out to Kafka
perfectly fine; but when it's included, it writes a couple of messages to
the sink then throws an abstract error:*
Caused by: org.apache.spark.util.TaskCompletionListenerException:
Self-suppression not permitted
at
org.apache.spark.TaskContextImpl.invokeListeners(TaskContextImpl.scala:145)
at
org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:124)
at org.apache.spark.scheduler.Task.run(Task.scala:143)
at
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:444)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
at
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:447)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
org.apache.spark.scheduler.TaskSetManager Task 1 in stage 1.0 failed 4
times; aborting
job"[org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec]
Data source write support
org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@7337dcc is
aborting."
org.apache.spark.SparkException: Writing job aborted.
at
org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.writeWithV2(WriteToDataS
ourceV2Exec.scala:413)
at
org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.writeWithV2$(WriteToData
SourceV2Exec.scala:361)
at
org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec.writeWithV2(Write
ToDataSourceV2Exec.scala:322)
at
org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec.run(WriteToDataSo
urceV2Exec.scala:329)
at
org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result$lzycompute(V2Command
Exec.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(V2CommandExe
c.scala:45)
at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3625)
at
org.apache.spark.sql.Dataset.$anonfun$collect$1(Dataset.scala:2938)
at
org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3616)
at
org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution
.scala:100)
at
org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:1
60)
at
org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution
.scala:87)
at
org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:763)
at
org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3614)
at org.apache.spark.sql.Dataset.collect(Dataset.scala:2938)
at
org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runBatch$16(Micro
BatchExecution.scala:576)
at
org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution
.scala:100)
at
org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:1
60)
at
org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution
.scala:87)
at
org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:763)
at
org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
at
org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runBatch$15(Micro
BatchExecution.scala:571)
*I have tried using with and without watermark, and various output modes,
all seems to result in the same error.
*
Any help would be greatly appreciated!
--
Sent from: http://apache-spark-user-list.1001560.n3.nabble.com/
---------------------------------------------------------------------
To unsubscribe e-mail: user-unsubscr...@spark.apache.org
