Tablestore:バッチコンピューティング

コードブロック

説明

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

Spark の DataFrameReader インターフェースを使用して Tablestore からデータを読み取り、DataFrame オブジェクトとしてロードします。

• format("tablestore") は、ServiceLoader メソッドを使用して Spark Tablestore コネクタがロードされることを指定します。 特定の構成については、META-INF.services ディレクトリを参照してください。

• instanceName、tableName、endpoint、accessKeyId、accessKeySecret は、それぞれ Tablestore インスタンス名、データテーブル名、インスタンスエンドポイント、Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID、Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey シークレットを指定します。

• split.size.mbs は、各分割のサイズを MB 単位で指定します。 デフォルト値: 100。このパラメータの値が小さいほど、より多くの分割が生成され、より多くの Spark タスクに対応します。

• .schema("salt LONG, UserId STRING, OrderId STRING, price DOUBLE, timestamp LONG") は、データテーブルのスキーマを指定します。 スキーマには、フィールド名とデータ型が指定されています。 サンプルデータテーブルには、salt (Long 型)、UserId (String 型)、OrderId (String 型)、price (Double 型)、timestamp (Long 型) の 5 つのフィールドがあります。

  説明

  最新バージョンでは、.schema() メソッドを使用してカタログ構成を置き換えることができます。 ビジネス要件に基づいてバージョンを選択してください。

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

JSON 文字列 dataCatalog を定義して、Tablestore のスキーマ情報を記述します。 各フィールドの名前とデータ型は、キーと値のペアを使用して指定されます。

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'' という条件を満たすレコードの総数をカウントします。