Can you write a script for watermarking?

Question

Assisting AI · Accepted Answer

Watermarking is a technique used in event‑time stream processing to handle late data. A watermark is a timestamp that indicates that the system has seen all events up to that point in event time. In a typical script, you would assign event timestamps to each record, then generate a watermark that lags behind the maximum event time by a configured delay (e.g., 5 minutes). In PySpark Structured Streaming you can do:

```python
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, expr

spark = SparkSession.builder.appName("WatermarkExample").getOrCreate()

stream = spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "topic")
    .load()

# Assume the value is JSON with a field "event_time"
parsed = stream.selectExpr("CAST(value AS STRING) as json")
    .selectExpr("json.*")

# Convert event_time to timestamp and set watermark
watermarked = parsed.withColumn("event_ts", col("event_time").cast("timestamp"))
    .withWatermark("event_ts", "5 minutes")

# Example aggregation
result = watermarked.groupBy("key").agg(expr("count(*) as cnt"))

query = result.writeStream.outputMode("update").format("console").start()
query.awaitTermination()
```

The `withWatermark` call tells Spark to drop state older than 5 minutes after the maximum event time seen. This ensures that late events are either processed or dropped based on the chosen policy. The time complexity is O(n) for processing each micro‑batch, and the memory overhead is proportional to the number of distinct keys retained in state.

Can you write a script for watermarking?

💡 Model Answer

🎤 Get questions like this answered in real-time