流计算

代码块

描述

val ordersDF = sparkSession.readStream
  .format("tablestore")
  .option("instance.name", instanceName)
  .option("table.name", tableName)
  .option("tunnel.id", tunnelId)
  .option("endpoint", endpoint)
  .option("access.key.id", accessKeyId)
  .option("access.key.secret", accessKeySecret)
  .option("maxoffsetsperchannel", maxOffsetsPerChannel) // default 10000
  .option("catalog", dataCatalog)
  .load()
  .createTempView("order_source_stream_view")

通过Spark Structured Streaming，从表格存储中读取流式数据，并将其加载为一个流式DataFrame，同时注册为临时视图order_source_stream_view。

• format("tablestore")表示使用ServiceLoader方式加载Spark Tablestore connector。具体配置，请参见META-INF.services目录。

• instanceName、tableName、tunnelId、endpoint、accessKeyId、accessKeySecret分别表示表格存储的实例名称、数据表名称、通道ID、实例endpoint、阿里云账号或RAM用户的AccessKey ID和AccessKey Secret。

• catalog是一个JSON串，用于描述表格存储中的Schema信息。

• maxoffsetsperchannel表示每一个mini-batch中每一个channel（分区）最多读取的数据量，默认值为10000。

val dataCatalog: String =
  s"""
     |{"columns": {
     |    "UserId": {"type":"string"},
     |    "OrderId": {"type":"string"},
     |    "price": {"type":"double"},
     |    "timestamp": {"type":"long"}
     | }
     |}""".stripMargin

定义了一个JSON字符串dataCatalog，用于描述表格存储中的Schema信息。每个字段的名称及其数据类型均以键值对的形式呈现。

val aggDF = sparkSession.sql(
  "SELECT CAST(window.start AS String) AS begin, CAST(window.end AS String) AS end, count(*) AS count, " +
    "CAST(sum(price) AS Double) AS totalPrice FROM order_source_stream_view " +
    "GROUP BY window(to_timestamp(timestamp / 1000), '30 seconds')")
    
val query = aggDF.writeStream
  .outputMode("complete")
  .format("console")
  .option("truncate", value = false)
  .option("checkpointLocation", checkpointLocation)
  .option("triggerInterval", 10000) // custom
  .start()

通过Spark SQL对流式数据进行时间窗口聚合分析，并将结果输出到控制台。

• SQL查询部分

  • 时间窗口：基于timestamp字段，将数据划分为30秒的时间窗口。

  • 聚合操作：统计每个窗口内的记录总数，并计算每个窗口内price字段的总和。

• 配置流式查询的输出方式

  • outputMode("complete")表示指定输出模式为complete，即每次触发时输出完整的聚合结果。

  • format("console")：表示将结果输出到控制台。

  • option("truncate", value = false)表示禁用字段截断，确保完整显示字段值。

  • option("checkpointLocation", checkpointLocation)表示设置检查点目录，用于容错和恢复。

  • option("triggerInterval", 10000)表示设置触发间隔为10秒，即每10秒执行一次聚合操作。

  • start()表示启动流式查询。