批计算

代码块

描述

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)
  // 最新版本支持使用Schema方式替换catalog的配置。
  .schema("salt LONG, UserId STRING, OrderId STRING, price DOUBLE, timestamp LONG")
  .load()

通过Spark的DataFrameReader接口，读取表格存储中的数据，并将其加载为一个DataFrame对象。

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

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

• split.size.mbs表示每个Split的切分大小，默认值为100，单位为MB。此值越小产生的Split会越多，对应Spark的Task也会越多。

• .schema("salt LONG, UserId STRING, OrderId STRING, price DOUBLE, timestamp LONG")表示指定数据表的Schema，定义字段名称及其数据类型。示例中的数据表有salt（Long类型）、UserId（String类型）、OrderId（String类型）、price（Double类型）和timestamp（Long类型）五个字段。

  说明

  最新版本支持使用Schema方式替换catalog的配置，请根据实际情况选择。

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

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

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

对DataFrame执行过滤操作，筛选出满足条件salt = 1 AND UserId = 'user_A'的数据，并显示前20条记录。

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

将DataFrame注册为临时视图search_view，并通过Spark SQL执行多个聚合查询，统计满足不同条件的记录数。

• SQL查询1：统计符合条件salt = 1 AND UserId = 'user_A'的记录总数。

• SQL查询2：统计符合条件salt = 1 AND UserId = 'user_A' AND OrderId = '00002664-9d8b-441b-bad7-845202f3b142'的记录总数。

• SQL查询3：统计符合条件salt = 1 AND UserId >= 'user_A' AND UserId < 'user_B''的记录总数。