Tablestore:Batch computing

Code block

Description

val df = sparkSession.read
  .format("tablestore")
  .option("instance.name", instanceName)
  .option("table.name", tableName)
  .option("endpoint", endpoint)
  .option("access.key.id", accessKeyId)
  .option("access.key.secret", accessKeySecret)
  .option("split.size.mbs", 100)
  // .option("catalog", dataCatalog)
  // The latest version allows you to use the .schema() method to replace the catalog configurations.
  .schema("salt LONG, UserId STRING, OrderId STRING, price DOUBLE, timestamp LONG")
  .load()

Reads data from Tablestore by using Spark's DataFrameReader interface and loads the data as a DataFrame object.

• 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, endpoint, accessKeyId, and accessKeySecret specify the Tablestore instance name, data table name, instance endpoint, AccessKey ID of the Alibaba Cloud account or RAM user, and AccessKey secret of the Alibaba Cloud account or RAM user, respectively.

• split.size.mbs specifies the size of each split in MB. Default value: 100. The smaller the value of this parameter, the more splits will be generated, corresponding to more Spark tasks.

• .schema("salt LONG, UserId STRING, OrderId STRING, price DOUBLE, timestamp LONG") specifies the schema of the data table. The field names and their data types are specified in the schema. The sample data table has five fields: salt (Long type), UserId (String type), OrderId (String type), price (Double type), and timestamp (Long type).

  Note

  The latest version allows you to use the .schema() method to replace the catalog configurations. Select a version based on your business requirements.

val dataCatalog: String =
  s"""
     |{"columns": {
     |    "salt": {"type":"long"},
     |    "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.

df.filter("salt = 1 AND UserId = 'user_A'").show(20, truncate = false)

Performs a filter operation on DataFrame to select data that meets the conditions salt = 1 AND UserId = 'user_A', and displays the first 20 records.

df.createTempView("search_view")
val searchDF = sparkSession.sql("SELECT COUNT(*) FROM search_view WHERE salt = 1 AND UserId = 'user_A'")
searchDF.show()
val searchDF2 = sparkSession.sql("SELECT COUNT(*) FROM search_view WHERE salt = 1 AND UserId = 'user_A'" +
  " AND OrderId = '00002664-9d8b-441b-bad7-845202f3b142'")
searchDF2.show()
val searchDF3 = sparkSession.sql("SELECT COUNT(*) FROM search_view WHERE salt = 1 AND UserId >= 'user_A' AND UserId < 'user_B'")
searchDF3.show()

Registers DataFrame as a temporary view search_view and executes multiple aggregate queries through Spark SQL to count records that meet different conditions.

• SQL query 1: Counts the total number of records that meet the conditions salt = 1 AND UserId = 'user_A'.

• SQL query 2: Counts the total number of records that meet the conditions salt = 1 AND UserId = 'user_A' AND OrderId = '00002664-9d8b-441b-bad7-845202f3b142'.

• SQL query 3: Counts the total number of records that meet the conditions salt = 1 AND UserId >= 'user_A' AND UserId < 'user_B''.