窗口函数与流式查询

在 Databricks 中使用 Spark SQL 进行数据转换

Disha Mukherjee

Lead Data Engineer

当 groupBy 不够用时

recraft: half: 放大镜覆盖电子表格行，部分计算值被高亮

 

  • groupBy() → 每组一行（总计、平均）
  • 窗口函数 → 新增列，保留每一行
  • 用途：滚动合计、排名、行间对比
在 Databricks 中使用 Spark SQL 进行数据转换

滚动合计

from pyspark.sql.window import Window

window_spec = (
    Window.partitionBy("Customer_ID")
    .orderBy("Date")
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)
)


df_running = df_valid.withColumn(
    "running_total",
    F.round(F.sum("Transaction_Amount").over(window_spec), 2)
)
在 Databricks 中使用 Spark SQL 进行数据转换

滚动合计

+-----------+-------------------+------------------+-------------+
|Customer_ID|Date               |Transaction_Amount|running_total|
+-----------+-------------------+------------------+-------------+
|CUST003    |2023-01-03 00:00:00|5752.36           |5752.36      |
|CUST003    |2023-02-14 00:00:00|12340.00          |18092.36     |
+-----------+-------------------+------------------+-------------+
在 Databricks 中使用 Spark SQL 进行数据转换

客户排名

customer_totals = (
    df_valid.groupBy("Customer_ID")
    .agg(F.round(F.sum("Transaction_Amount"), 2).alias("total_revenue"))
)


rank_window = Window.orderBy(F.col("total_revenue").desc())

df_ranked = customer_totals.withColumn("revenue_rank", F.rank().over(rank_window))

+-----------+-------------+------------+
|Customer_ID|total_revenue|revenue_rank|
+-----------+-------------+------------+
|CUST1469   |99996.20     |1           |
|CUST19129  |99990.14     |2           |
|CUST39417  |99982.21     |3           |
+-----------+-------------+------------+
在 Databricks 中使用 Spark SQL 进行数据转换

什么是流处理？

 

  • 批处理：一次性处理固定数据集
  • 流处理：数据到达即增量处理
  • 适用于交易流、日志、事件数据

recraft: half: 数据持续以流的形式流入处理引擎，带有指示方向的箭头

在 Databricks 中使用 Spark SQL 进行数据转换

基于文件的流处理

$$

Stream directory:
  day_1.csv  (108 KB)
  day_2.csv  (108 KB)
  day_3.csv  (108 KB)
  day_4.csv  (108 KB)
  day_5.csv  (108 KB)

 

  • CSV 文件目录 = 流式数据源
  • 每个新文件 = 一个微批
  • 新文件自动被读取
  • 必须显式定义模式
在 Databricks 中使用 Spark SQL 进行数据转换

读取流

df_stream = (
    spark.readStream.format("csv")
    .option("header", "true")
    .schema(streaming_schema)
    .load(STREAM_DIR)
)

print(df_stream.isStreaming)

True
在 Databricks 中使用 Spark SQL 进行数据转换

什么是检查点？

recraft: half: 流水线中的书签/保存点标记，表示进度已保存

 

$$

  • Checkpoint = 磁盘上的元数据目录
  • 记录已处理的文件
  • 重启后 Spark 从断点续跑
  • 防止重复处理
在 Databricks 中使用 Spark SQL 进行数据转换

写出流

query = (
    df_stream.writeStream.format("delta")
    .outputMode("append")
    .option("checkpointLocation", CHECKPOINT_DIR)
    .option("path", DELTA_PATH)
    .trigger(availableNow=True)
    .start()
)
query.awaitTermination()

Status:       Stopped
Rows written: 5,000
在 Databricks 中使用 Spark SQL 进行数据转换

监控：status 与 lastProgress

print(query.status)

progress = query.lastProgress
print(f"Rows processed: {progress['numInputRows']}")
print(f"Rows/sec: {progress['processedRowsPerSecond']:.0f}")

{'message': 'Stopped', 'isDataAvailable': False, 'isTriggerActive': False}
Rows processed: 5,000
Rows/sec:       752
在 Databricks 中使用 Spark SQL 进行数据转换

检查点恢复

query_restart = (
    df_stream.writeStream.format("delta")
    .option("checkpointLocation", CHECKPOINT_DIR)
    .option("path", DELTA_PATH)
    .trigger(availableNow=True)
    .start()
)
query_restart.awaitTermination()

Rows on restart: 0
在 Databricks 中使用 Spark SQL 进行数据转换

让我们一起练习吧！

在 Databricks 中使用 Spark SQL 进行数据转换

Preparing Video For Download...