Tablestore：Realtime Compute

代碼塊

描述

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，從Table Store中讀取流式資料，並將其載入為一個流式DataFrame，同時註冊為臨時視圖order_source_stream_view。

• format("tablestore")表示使用ServiceLoader方式載入Spark Tablestore connector。具體配置，請參見META-INF.services目錄。

• instanceName、tableName、tunnelId、endpoint、accessKeyId、accessKeySecret分別表示Table Store的執行個體名稱、資料表名稱、通道ID、執行個體endpoint、阿里雲帳號或RAM使用者的AccessKey ID和AccessKey Secret。

• catalog是一個JSON串，用於描述Table Store中的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，用於描述Table Store中的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()表示啟動流式查詢。