readStream

schema = "device STRING, ecommerce STRUCT<purchase_revenue_in_usd: DOUBLE, total_item_quantity: BIGINT, unique_items: BIGINT>, event_name STRING, event_previous_timestamp BIGINT, event_timestamp BIGINT, geo STRUCT<city: STRING, state: STRING>, items ARRAY<STRUCT<coupon: STRING, item_id: STRING, item_name: STRING, item_revenue_in_usd: DOUBLE, price_in_usd: DOUBLE, quantity: BIGINT>>, traffic_source STRING, user_first_touch_timestamp BIGINT, user_id STRING"

df = (spark.readStream
  .schema(schema)
  .option("maxFilesPerTrigger", 1)
  .parquet(eventsPath)
)

Build streaming DataFrames

df.isStreaming #  Judge DataFrame Whether it is Streaming, Return here True

from pyspark.sql.functions import col, approx_count_distinct, count

emailTrafficDF = (df.filter(col("traffic_source") == "email")
  .withColumn("mobile", col("device").isin(["iOS", "Android"]))
  .select("user_id", "event_timestamp", "mobile")
)

Write streaming query results

checkpointPath = f"{
      userhome}/email_traffic/checkpoint"
outputPath = f"{
      userhome}/email_traffic/output"

devicesQuery = (emailTrafficDF.writeStream
  .outputMode("append")
  .format("parquet")
  .queryName("email_traffic_p")
  .trigger(processingTime="1 second")
  .option("checkpointLocation", checkpointPath)
  .start(outputPath)
)

Monitor streaming query

Get ID of streaming query

devicesQuery.id

Get status of streaming query

devicesQuery.status

devicesQuery.awaitTermination(5)

Stop the streaming query

devicesQuery.stop()