Tablestore:Stream processing

Code block

Description

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")

Reads streaming data from Tablestore by using Spark Structured Streaming, loads the streaming data as a streaming DataFrame, and registers the streaming DataFrame as a temporary view order_source_stream_view.

• format("tablestore") specifies that the Spark Tablestore connector is loaded by using the ServiceLoader method. For specific configurations, see the META-INF.services directory.

• instanceName, tableName, tunnelId, endpoint, accessKeyId, and accessKeySecret specify the Tablestore instance name, data table name, tunnel ID, instance endpoint, AccessKey ID of the Alibaba Cloud account or RAM user, and AccessKey secret of the Alibaba Cloud account or RAM user, respectively.

• catalog is a JSON string used to describe the schema information in Tablestore.

• maxoffsetsperchannel specifies the maximum amount of data read from each channel in each mini-batch. The default value is 10000.

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

Defines a JSON string dataCatalog to describe the schema information in Tablestore. The name and data type of each field are specified by using a key-value pair.

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()

Aggregates and analyzes streaming data based on a specific time window by using Spark SQL and displays the results in the Tablestore console.

• SQL query

  • Time window: Divides data based on a 30-second time windows specified by the timestamp field.

  • Aggregate operation: Counts the total number of records in each window and calculates the sum of the values in the price field in each window.

• Configure the output method for streaming query

  • outputMode("complete") specifies that the output mode is complete, which means that the complete aggregation result is displayed each time the query is triggered.

  • format("console") specifies that the results are displayed in the Tablestore console.

  • option("truncate", value = false) specifies that field truncation is disabled to ensure that full field values are displayed.

  • option("checkpointLocation", checkpointLocation) specifies that the checkpoint directory is configured for fault tolerance and recovery.

  • option("triggerInterval", 10000) specifies that the trigger interval is set to 10 seconds, which means that the aggregation operation is performed every 10 seconds.

  • start() specifies that streaming query is started.