[SPARK-19903][PYSPARK][SS] window operator miss the watermark metadata of time column

[uncleGen]

What changes were proposed in this pull request?

reproduce code:

import sys
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split, window
bootstrapServers = sys.argv[1]
subscribeType = sys.argv[2]
topics = sys.argv[3]
spark = SparkSession\
  .builder\
  .appName("StructuredKafkaWordCount")\
  .getOrCreate()

lines = spark\
  .readStream\
  .format("kafka")\
  .option("kafka.bootstrap.servers", bootstrapServers)\
  .option(subscribeType, topics)\
  .load()\
  .selectExpr("CAST(value AS STRING)", "CAST(timestamp AS TIMESTAMP)")

words = lines.select(explode(split(lines.value, ' ')).alias('word'),lines.timestamp)

windowedCounts = words.withWatermark("timestamp", "30 seconds").groupBy(
window(words.timestamp, "30 seconds", "30 seconds"), words.word
).count()

query = windowedCounts\
  .writeStream\
  .outputMode('append')\
  .format('console')\ 
  .option("truncate", "false")\
  .start()
query.awaitTermination()

An exception was thrown:

pyspark.sql.utils.AnalysisException: Append output mode not supported when there are streaming aggregations on streaming DataFrames/DataSets without watermark;;
Aggregate [window#32, word#21], [window#32 AS window#26, word#21, count(1) AS count#31L]
+- Filter ((timestamp#16 >= window#32.start) && (timestamp#16 < window#32.end))
   +- Expand [ArrayBuffer(named_struct(start, ...]
      +- EventTimeWatermark timestamp#16: timestamp, interval 10 seconds
         +- Project [word#21, timestamp#16]
            +- Generate explode(split(value#15,  )), true, false, [word#21]
               +- Project [cast(value#1 as string) AS value#15, cast(timestamp#5 as timestamp) AS timestamp#16]
                  +- StreamingRelation DataSource(org.apache.spark.sql.SparkSession ...]

IIUC, the root cause is: words.withWatermark("timestamp", "30 seconds") add the watermark metadata into time column, but in groupBy( window(words.timestamp, "30 seconds", "30 seconds"), words.word ), the words.timestamp miss the metadata. At last, it failed to pass the check:

use @viirya 's more clear explanation:

For now, after withWatermark, we only update the metadata for the column of event time. The expression id is the same. So once we use the column before adding watermark words.timestamp as grouping expression, it binds to the old attribute before watermarking.

if (watermarkAttributes.isEmpty) {
      throwError(
            s"$outputMode output mode not supported when there are streaming aggregations on " +
                s"streaming DataFrames/DataSets without watermark")(plan)
}

In this pr, pass a UnresolvedAttribute to window instead of a Column

How was this patch tested?

Jenkins

[viirya]

hmm, doesn't this break the current API?

[uncleGen]

IIUC, it is OK for current codebase. Am I missing something?

[viirya]

After this change, you can't pass in a Column. But it is supported for now.

[uncleGen]

@viirya Sounds reasonable, I pushed an update, take a review please.

[SparkQA]

Test build #74959 has finished for PR 17371 at commit 654c512.

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

[SparkQA]

Test build #74967 has finished for PR 17371 at commit 890c6e6.

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

[viirya]

For now, after withWatermark, we only update the metadata for the column of event time. The expression id is the same. So once we use the column before adding watermark words.timestamp as grouping expression, it binds to the old attribute before watermarking.

I am thinking, should we create new expression id for the watermarking column with withWatermark? So we must write the query like:

wordsWithWatermark = words.withWatermark("timestamp", "30 seconds")
windowedCounts = wordsWithWatermark.groupBy(window(wordsWithWatermark.timestamp, "30 seconds", "30 seconds"), wordsWithWatermark.word).count()

[uncleGen]

@viirya Great, you give a more clear explanation.

I am thinking, should we create new expression id for the watermarking column with withWatermark? So we must write the query like:

It really can fix this problem, but not very user-friendly.

[viirya]

IMHO, the output after withWatermark should be new attribute and have new expression id. Maybe @zsxwing @marmbrus have more insights on this?

Btw, does this issue also happen in Scala code?

[marmbrus]

I really think the core problem here is that we allow you to use resolved attributes at all in the user API. Unfortunately we are somewhat stuck with that bad decision. Personally, I never use df['col'] and only ever use col("col") since that avoids the problem.

However, I don't think that piecemeal switching to unresolved attributes is a good idea.

[viirya]

Unfortunately, yes, allowing resolved attributes in user API will have this kind of trouble.

However, I don't think that piecemeal switching to unresolved attributes is a good idea.

Agreed. Should we create new attributes after withWatermark to avoid the problem? It might be cumbersome from the user side, however.

[marmbrus]

I don't think that will solve the problem though. You will just get a different error message.

[viirya]

yeah, I just tried it. IncrementalExecution will re-new the attribute for each batch. Although we can replace the attribute...

[marmbrus]

Can we add an analysis rule that just pulls up missing metadata from attributes in the child? It could run once after other rules.

[HyukjinKwon]

gentle ping @uncleGen, is this PR still active?