[SPARK-36314][SS] Update Sessionization examples to use native support of session window

[HeartSaVioR]

What changes were proposed in this pull request?

This PR proposes to update Sessionization examples to use native support of session window. It also adds the example for PySpark as native support of session window is available to PySpark as well.

Why are the changes needed?

We should guide the simplest way to achieve the same workload. I'll provide another example for cases we can't do with native support of session window.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Manually tested.

[HeartSaVioR]

Target branches are master/3.2.

[HeartSaVioR]

cc. @viirya @xuanyuanking

[SparkQA]

Test build #141737 has finished for PR 33548 at commit c62ea25.

• This patch fails Python style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #141741 has finished for PR 33548 at commit 039f53a.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46250/

[viirya]

structured_sessionization.py?

[viirya]

Looks good. One typo found.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46254/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46254/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46250/

[SparkQA]

Test build #141744 has finished for PR 33548 at commit f6cf587.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[xuanyuanking]

LGTM

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46257/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46257/

[viirya]

Thanks! Merging to master/3.2.

[HeartSaVioR]

Thanks all for the quick reviews and merge!

[thenooz]

Hello fellow spark structured streaming sessionizers. According to the apache spark docs, output mode update is not allowed with session window. I tried running the example code in my local machine for 3.2.1 and 3.2.0 and it failed with error:

org.apache.spark.sql.AnalysisException: Update output mode not supported for session window on streaming DataFrames/DataSets;

According to this example code it can be done. What am I missing?

[HeartSaVioR]

Ah we seem to miss updating the example... We allowed update mode first and disallowed due to unclear semantic.

Could you please raise a PR to update the output mode to append for Scala/Java/Python examples for session window if you don't mind? I can jump in if you're not available to address this. Thanks in advance!

[thenooz]

Sure, but you see, if we do it with append mode, then that means the sessions will be returned only after they are closed. Due to this, possibility and purpose of real-time session tagging is defeated. Should we not revert the example to mapGroupsWithState method?

[HeartSaVioR]

We still have an example of session window against (flat)mapGroupsWithState. We just don't want to duplicate the code example.

The main reason I proposed dropping the functionality of update mode for session window was due to the characteristic of session window. When someone uses update mode for the streaming query, it does not only mean they want to get update immediately, but also mean they are going to "upsert" the target table.

Upsertion is normally performed by replacing the existing value(s) for specific key to new value(s). If we consider the out-of-order events, session window will expand arbitrarily - both start and end of session can change. (Technically saying, the old example of session window was incorrect in terms of event-time semantic.) Given Spark does not produce the old value(s), the query can't always safely upsert the result. E.g. user A has multiple sessions in the target table and some of sessions were already closed. Spark never gives them, and the query is in a risk of dropping these old-but-valid sessions.

[thenooz]

Got it. So the problem is we run a risk of taking wrong real-time decisions on the basis of immediate updates and it is very much possible the session we calculated was incorrect due to late events. So, the safer and more accurate solution would be append mode with watermarking.
But still, back to a probably annoying question: Can we allow update mode with session window with a warning? You know, for people who take big risks.
Also, will check on updating the example code. I am new to this.
Thank you very much.