[
https://issues.apache.org/jira/browse/SPARK-30634?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dongjoon Hyun resolved SPARK-30634.
-----------------------------------
Resolution: Invalid
Hi, [~yurkao]. Sorry, but JIRA is not for Q&A. Please send to dev mailing list
for the question.
> Delta Merge and Arbitrary Stateful Processing in Structured streaming
> (foreachBatch)
> -------------------------------------------------------------------------------------
>
> Key: SPARK-30634
> URL: https://issues.apache.org/jira/browse/SPARK-30634
> Project: Spark
> Issue Type: Question
> Components: Examples, Spark Core, Structured Streaming
> Affects Versions: 2.4.3
> Environment: Spark 2.4.3 (scala 2.11.12)
> Delta: 0.5.0
> Java(TM) SE Runtime Environment (build 1.8.0_91-b14)
> OS: Ubuntu 18.04 LTS
>
> Reporter: Yurii Oleynikov
> Priority: Trivial
> Attachments: Capture1.PNG
>
>
> Hi ,
> I have an application that makes Arbitrary Stateful Processing in Structured
> Streaming and used delta.merge to update delta table and faced strange
> behaviour:
> 1. I've noticed that logs inside implementation of
> {{MapGroupsWithStateFunction}}/ {{FlatMapGroupsWithStateFunction}} in my
> application outputted twice.
> 2. While finding a root cause I've also found that number State rows reported
> by Spark is also doubles.
>
> I thought that may be there's a bug in my code, so I back to
> {{JavaStructuredSessionization}} from Apache Spark examples and changed it a
> bit. Still got same result.
> The problem happens only if I do not perform datch.DF.persist inside
> foreachBatch.
> {code:java}
> StreamingQuery query = sessionUpdates
> .writeStream()
> .outputMode("update")
> .foreachBatch((VoidFunction2<Dataset<SessionUpdate>, Long>) (batchDf,
> v2) -> {
> // following doubles number of spark state rows and causes
> MapGroupsWithStateFunction to log twice withport persisting
> deltaTable.as("sessions").merge(batchDf.toDF().as("updates"),
> mergeExpr)
> .whenNotMatched().insertAll()
> .whenMatched()
> .updateAll()
> .execute();
> })
> .trigger(Trigger.ProcessingTime(10000))
> .queryName("ACME")
> .start();
> {code}
> According to
> [https://docs.databricks.com/_static/notebooks/merge-in-streaming.html] and
> [Apache spark
> docs|https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#using-foreach-and-foreachbatch]
> there's seems to be no need to persist dataset/dataframe inside
> {{foreachBatch.}}
> Sample code from Apache Spark examples with delta:
> [JavaStructuredSessionization with Delta
> merge|https://github.com/yurkao/delta-merge-sss/blob/master/src/main/java/JavaStructuredSessionization.java]
>
>
> Appreciate your clarification.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
