Spark コネクタを使用して StarRocks からデータを読み取る
StarRocks は、Apache Spark™ 用に開発したコネクタである StarRocks Connector for Apache Spark™（以下、Spark コネクタ）を提供しており、これを使用して StarRocks テーブルからデータを読み取ることができます。Spark を使用して、StarRocks から読み取ったデータに対して複雑な処理や機械学習を行うことができます。
Spark コネクタは、Spark SQL、Spark DataFrame、Spark RDD の3つの読み取り方法をサポートしています。
Spark SQL を使用して StarRocks テーブルに一時ビューを作成し、その一時ビューを使用して StarRocks テーブルから直接データを読み取ることができます。
また、StarRocks テーブルを Spark DataFrame または Spark RDD にマッピングし、その Spark DataFrame または Spark RDD からデータを読み取ることもできます。Spark DataFrame の使用を推奨します。
注意
StarRocks テーブルの SELECT 権限を持つユーザーのみが、このテーブルからデータを読み取ることができます。GRANT の指示に従って、ユーザーに権限を付与することができます。
使用上の注意
- データを読み取る前に StarRocks 上でデータをフィルタリングすることで、転送されるデータ量を削減できます。
- データ読み取りのオーバーヘッドが大きい場合、適切なテーブル設計とフィルタ条件を使用して、Spark が一度に過剰なデータを読み取らないようにすることができます。これにより、ディスクやネットワーク接続への I/O 圧力を軽減し、通常のクエリが適切に実行されることを保証できます。
バージョン要件
| Spark コネクタ | Spark | StarRocks | Java | Scala |
|---|---|---|---|---|
| 1.1.2 | 3.2, 3.3, 3.4, 3.5 | 2.5 and later | 8 | 2.12 |
| 1.1.1 | 3.2, 3.3, 3.4 | 2.5 and later | 8 | 2.12 |
| 1.1.0 | 3.2, 3.3, 3.4 | 2.5 and later | 8 | 2.12 |
| 1.0.0 | 3.x | 1.18 and later | 8 | 2.12 |
| 1.0.0 | 2.x | 1.18 and later | 8 | 2.11 |
注意
- 異なるコネクタバージョン間の動作変更については、Upgrade Spark connector を参照してください。
- バージョン 1.1.1 以降、コネクタは MySQL JDBC ドライバを提供していないため、Spark クラスパスに手動でドライバをインポートする必要があります。ドライバは Maven Central で見つけることができます。
- バージョン 1.0.0 では、Spark コネクタは StarRocks からのデータ読み取りのみをサポートしています。バージョン 1.1.0 以降では、StarRocks からのデータ読み取りと書き込みの両方をサポートしています。
- バージョン 1.0.0 と 1.1.0 の間では、パラメータとデータ型のマッピングが異なります。Upgrade Spark connector を参照してください。
- 一般的な場合、バージョン 1.0.0 に新しい機能は追加されません。できるだけ早く Spark コネクタをアップグレードすることをお勧めします。
Spark コネクタの取得
ビジネスニーズに合った Spark コネクタ .jar パッケージを取得するには、次のいずれかの方法を使用します。
- コンパイル済みパッケージをダウンロードする。
- Maven を使用して Spark コネクタに必要な依存関係を追加する。（この方法は Spark コネクタ 1.1.0 以降でのみサポートされています。）
- 手動でパッケージをコンパイルする。
Spark コネクタ 1.1.0 以降
Spark コネクタ .jar パッケージは、次の形式で命名されています。
starrocks-spark-connector-${spark_version}_${scala_version}-${connector_version}.jar
たとえば、Spark 3.2 と Scala 2.12 で Spark コネクタ 1.1.0 を使用したい場合、starrocks-spark-connector-3.2_2.12-1.1.0.jar を選択できます。
注意
通常、最新の Spark コネクタバージョンは、最新の 3 つの Spark バージョンと互換性があります。
コンパイル済みパッケージをダウンロードする
さまざまなバージョンの Spark コネクタ .jar パッケージは Maven Central Repository で入手できます。
Maven 依存関係を追加する
Spark コネクタに必要な依存関係を次のように設定します。
注意
spark_version、scala_version、connector_versionを使用する Spark バージョン、Scala バージョン、および Spark コネクタバージョンに置き換える必要があります。
<dependency>
<groupId>com.starrocks</groupId>
<artifactId>starrocks-spark-connector-${spark_version}_${scala_version}</artifactId>
<version>${connector_version}</version>
</dependency>
たとえば、Spark 3.2 と Scala 2.12 で Spark コネクタ 1.1.0 を使用したい場合、依存関係を次のように設定します。
<dependency>
<groupId>com.starrocks</groupId>
<artifactId>starrocks-spark-connector-3.2_2.12</artifactId>
<version>1.1.0</version>
</dependency>
手動でパッケージをコンパイルする
-
Spark コネクタのコード をダウンロードします。
-
次のコマンドを使用して Spark コネクタをコンパイルします。
注意
spark_versionを使用する Spark バージョンに置き換える必要があります。sh build.sh <spark_version>たとえば、Spark 3.2 で Spark コネクタを使用したい場合、次のように Spark コネクタをコンパイルします。
sh build.sh 3.2 -
target/パスに移動し、コンパイル時に生成されたstarrocks-spark-connector-3.2_2.12-1.1.0-SNAPSHOT.jarのような Spark コネクタ .jar パッケージを確認します。注意
公式にリリースされていない Spark コネクタバージョンを使用している場合、生成された Spark コネクタ .jar パッケージの名前には
SNAPSHOTがサフィックスとして含まれます。
Spark コネクタ 1.0.0
コンパイル済みパッケージをダウンロードする
手動でパッケージをコンパイルする
-
Spark コネクタのコード をダウンロードします。
注意
spark-1.0に切り替える必要があります。 -
Spark コネクタをコンパイルするために次のいずれかのアクションを実行します。
-
Spark 2.x を使用している場合、次のコマンドを実行します。デフォルトで Spark 2.3.4 に適した Spark コネクタをコンパイルします。
sh build.sh 2 -
Spark 3.x を使用している場合、次のコマンドを実行します。デフォルトで Spark 3.1.2 に適した Spark コネクタをコンパイルします。
sh build.sh 3
-
-
output/パスに移動し、コンパイル時に生成されたstarrocks-spark2_2.11-1.0.0.jarファイルを確認します。その後、ファイルを Spark のクラスパスにコピーします。- Spark クラスターが
Localモードで実行されている場合、ファイルをjars/パスに配置します。 - Spark クラスターが
Yarnモードで実行されている場合、ファイルを事前デプロイメントパッケージに配置します。
- Spark クラスターが
指定された場所にファイルを配置した後にのみ、Spark コネクタを使用して StarRocks からデータを読み取ることができます。
パラメータ
このセクションでは、Spark コネクタを使用して StarRocks からデータを読み取る際に設定する必要があるパラメータについて説明します。
共通パラメータ
次のパラメータは、Spark SQL、Spark DataFrame、Spark RDD の3つの読み取り方法すべてに適用されます。
| パラメータ | デフォルト値 | 説明 |
|---|---|---|
| starrocks.fenodes | None | StarRocks クラスター内の FE の HTTP URL。形式 <fe_host>:<fe_http_port>。複数の URL を指定する場合は、カンマ (,) で区切る必要があります。 |
| starrocks.table.identifier | None | StarRocks テーブルの名前。形式: <database_name>.<table_name>。 |
| starrocks.request.retries | 3 | Spark が StarRocks に読み取りリクエストを再送信できる最大回数。 |
| starrocks.request.connect.timeout.ms | 30000 | StarRocks に送信された読み取りリクエストがタイムアウトするまでの最大時間。 |
| starrocks.request.read.timeout.ms | 30000 | StarRocks に送信されたリクエストの読み取りがタイムアウトするまでの最大時間。 |
| starrocks.request.query.timeout.s | 3600 | StarRocks からのデータクエリがタイムアウトするまでの最大時間。デフォルトのタイムアウト期間は1時間です。-1 はタイムアウト期間が指定されていないことを意味します。 |
| starrocks.request.tablet.size | Integer.MAX_VALUE | 各 Spark RDD パーティションにグループ化される StarRocks タブレットの数。このパラメータの値が小さいほど、生成される Spark RDD パーティションの数が多くなります。Spark の並行性が高くなる一方で、StarRocks への負荷も増加します。 |
| starrocks.batch.size | 4096 | 一度に BEs から読み取ることができる最大行数。このパラメータの値を増やすことで、Spark と StarRocks 間で確立される接続数を減らし、ネットワーク遅延による余分な時間のオーバーヘッドを軽減できます。 |
| starrocks.exec.mem.limit | 2147483648 | クエリごとに許可される最大メモリ量。単位: バイト。デフォルトのメモリ制限は2 GBです。 |
| starrocks.deserialize.arrow.async | false | Arrow メモリ形式を Spark コネクタのイテレーションに必要な RowBatches に非同期で変換するかどうかを指定します。 |
| starrocks.deserialize.queue.size | 64 | Arrow メモリ形式を RowBatches に非同期で変換するタスクを保持する内部キューのサイズ。このパラメータは starrocks.deserialize.arrow.async が true に設定されている場合に有効です。 |
| starrocks.filter.query | None | StarRocks 上でデータをフィルタリングするための条件。複数のフィルタ条件を指定する場合は、and で結合する必要があります。StarRocks は、指定されたフィルタ条件に基づいて StarRocks テーブルからデータをフィルタリングし、その後 Spark によってデータが読み取られます。 |
| starrocks.timezone | JVM のデフォルトタイムゾーン | 1.1.1 以降でサポートされています。StarRocks の DATETIME を Spark の TimestampType に変換するために使用されるタイムゾーン。デフォルトは ZoneId#systemDefault() によって返される JVM のタイムゾーンです。形式は Asia/Shanghai のようなタイムゾーン名、または +08:00 のようなゾーンオフセットです。 |
Spark SQL と Spark DataFrame 用のパラメータ
次のパラメータは、Spark SQL と Spark DataFrame の読み取り方法にのみ適用されます。
| パラメータ | デフォルト値 | 説明 |
|---|---|---|
| starrocks.fe.http.url | None | FE の HTTP IP アドレス。このパラメータは Spark コネクタ 1.1.0 以降でサポートされています。このパラメータは starrocks.fenodes と同等です。どちらか一方を設定するだけで済みます。Spark コネクタ 1.1.0 以降では、starrocks.fe.http.url を使用することをお勧めします。starrocks.fenodes は廃止される可能性があります。 |
| starrocks.fe.jdbc.url | None | FE の MySQL サーバーに接続するために使用されるアドレス。形式: jdbc:mysql://<fe_host>:<fe_query_port>。注意 Spark コネクタ 1.1.0 以降では、このパラメータは必須です。 |
| user | None | StarRocks クラスターアカウントのユーザー名。ユーザーは StarRocks テーブルの SELECT 権限 を持っている必要があります。 |
| starrocks.user | None | StarRocks クラスターアカウントのユーザー名。このパラメータは Spark コネクタ 1.1.0 以降でサポートされています。このパラメータは user と同等です。どちらか一方を設定するだけで済みます。Spark コネクタ 1.1.0 以降では、starrocks.user を使用することをお勧めします。user は廃止される可能性があります。 |
| password | None | StarRocks クラスターアカウントのパスワード。 |
| starrocks.password | None | StarRocks クラスターアカウントのパスワード。このパラメータは Spark コネクタ 1.1.0 以降でサポートされています。このパラメータは password と同等です。どちらか一方を設定するだけで済みます。Spark コネクタ 1.1.0 以降では、starrocks.password を使用することをお勧めします。password は廃止される可能性があります。 |
| starrocks.filter.query.in.max.count | 100 | プレディケートプッシュダウン中に IN 式でサポートされる最大値数。IN 式で指定された値の数がこの制限を超える場合、IN 式で指定されたフィルタ条件は Spark 上で処理されます。 |
Spark RDD 用のパラメータ
次のパラメータは、Spark RDD の読み取り方法にのみ適用されます。
| パラメータ | デフォルト値 | 説明 |
|---|---|---|
| starrocks.request.auth.user | None | StarRocks クラスターアカウントのユーザー名。 |
| starrocks.request.auth.password | None | StarRocks クラスターアカウントのパスワード。 |
| starrocks.read.field | None | データを読み取りたい StarRocks テーブルの列。複数の列を指定する場合は、カンマ (,) で区切る必要があります。 |
StarRocks と Spark 間のデータ型マッピング
Spark コネクタ 1.1.0 以降
| StarRocks データ型 | Spark データ型 |
|---|---|
| BOOLEAN | DataTypes.BooleanType |
| TINYINT | DataTypes.ByteType |
| SMALLINT | DataTypes.ShortType |
| INT | DataTypes.IntegerType |
| BIGINT | DataTypes.LongType |
| LARGEINT | DataTypes.StringType |
| FLOAT | DataTypes.FloatType |
| DOUBLE | DataTypes.DoubleType |
| DECIMAL | DecimalType |
| CHAR | DataTypes.StringType |
| VARCHAR | DataTypes.StringType |
| STRING | DataTypes.StringType |
| DATE | DataTypes.DateType |
| DATETIME | DataTypes.TimestampType |
| JSON | DataTypes.StringType 注意: このデータ型マッピングは Spark コネクタ v1.1.2 以降でサポートされており、StarRocks バージョン 2.5.13、3.0.3、3.1.0 以降が必要です。 |
| ARRAY | サポートされていないデータ型 |
| HLL | サポートされていないデータ型 |
| BITMAP | サポートされていないデータ型 |
Spark コネクタ 1.0.0
| StarRocks データ型 | Spark データ型 |
|---|---|
| BOOLEAN | DataTypes.BooleanType |
| TINYINT | DataTypes.ByteType |
| SMALLINT | DataTypes.ShortType |
| INT | DataTypes.IntegerType |
| BIGINT | DataTypes.LongType |
| LARGEINT | DataTypes.StringType |
| FLOAT | DataTypes.FloatType |
| DOUBLE | DataTypes.DoubleType |
| DECIMAL | DecimalType |
| CHAR | DataTypes.StringType |
| VARCHAR | DataTypes.StringType |
| DATE | DataTypes.StringType |
| DATETIME | DataTypes.StringType |
| ARRAY | サポートされていないデータ型 |
| HLL | サポートされていないデータ型 |
| BITMAP | サポートされていないデータ型 |
StarRocks が使用する基盤ストレージエンジンの処理ロジックは、DATE および DATETIME データ型を直接使用する場合に期待される時間範囲をカバーできません。そのため、Spark コネクタは StarRocks の DATE および DATETIME データ型を Spark の STRING データ型にマッピングし、StarRocks から読み取った日付と時刻データに一致する読みやすい文字列テキストを生成します。
Spark コネクタのアップグレード
バージョン 1.0.0 からバージョン 1.1.0 へのアップグレード
-
バージョン 1.1.1 以降、Spark コネクタは MySQL の公式 JDBC ドライバである
mysql-connector-javaを提供していません。これは、mysql-connector-javaが使用する GPL ライセンスの制限によるものです。ただし、Spark コネクタはテーブルメタデータにアクセスするためにmysql-connector-javaを必要とするため、Spark クラスパスに手動でドライバを追加する必要があります。ドライバは MySQL サイト または Maven Central で見つけることができます。 -
バージョン 1.1.0 では、Spark コネクタは StarRocks にアクセスしてより詳細なテーブル情報を取得するために JDBC を使用します。そのため、
starrocks.fe.jdbc.urlを設定する必要があります。 -
バージョン 1.1.0 では、一部のパラメータが改名されました。現在、古いパラメータと新しいパラメータの両方が保持されています。等価なパラメータのペアのうち、どちらか一方を設定するだけで済みますが、古いパラメータは廃止される可能性があるため、新しいパラメータを使用することをお勧めします。
starrocks.fenodesはstarrocks.fe.http.urlに改名されました。userはstarrocks.userに改名されました。passwordはstarrocks.passwordに改名されました。
-
バージョン 1.1.0 では、Spark 3.x に基づいて一部のデータ型のマッピングが調整されました。
- StarRocks の
DATEは Spark のDataTypes.DateType（元はDataTypes.StringType）にマッピングされます。 - StarRocks の
DATETIMEは Spark のDataTypes.TimestampType（元はDataTypes.StringType）にマッピングされます。
- StarRocks の
例
以下の例では、StarRocks クラスターに test という名前のデータベースを作成し、ユーザー root の権限を持っていると仮定しています。例のパラメータ設定は Spark Connector 1.1.0 に基づいています。
ネットワーク設定
Spark が配置されているマシンが、StarRocks クラスターの FE ノードに http_port（デフォルト: 8030）および query_port（デフォルト: 9030）を介してアクセスでき、BE ノードに be_port（デフォルト: 9060）を介してアクセスできることを確認してください。
データ例
サンプルテーブルを準備するには、次の手順を実行します。
-
testデータベースに移動し、score_boardという名前のテーブルを作成します。MySQL [test]> CREATE TABLE `score_board`
(
`id` int(11) NOT NULL COMMENT "",
`name` varchar(65533) NULL DEFAULT "" COMMENT "",
`score` int(11) NOT NULL DEFAULT "0" COMMENT ""
)
ENGINE=OLAP
PRIMARY KEY(`id`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`id`)
PROPERTIES (
"replication_num" = "3"
); -
score_boardテーブルにデータを挿入します。MySQL [test]> INSERT INTO score_board
VALUES
(1, 'Bob', 21),
(2, 'Stan', 21),
(3, 'Sam', 22),
(4, 'Tony', 22),
(5, 'Alice', 22),
(6, 'Lucy', 23),
(7, 'Polly', 23),
(8, 'Tom', 23),
(9, 'Rose', 24),
(10, 'Jerry', 24),
(11, 'Jason', 24),
(12, 'Lily', 25),
(13, 'Stephen', 25),
(14, 'David', 25),
(15, 'Eddie', 26),
(16, 'Kate', 27),
(17, 'Cathy', 27),
(18, 'Judy', 27),
(19, 'Julia', 28),
(20, 'Robert', 28),
(21, 'Jack', 29); -
score_boardテーブルをクエリします。MySQL [test]> SELECT * FROM score_board;
+------+---------+-------+
| id | name | score |
+------+---------+-------+
| 1 | Bob | 21 |
| 2 | Stan | 21 |
| 3 | Sam | 22 |
| 4 | Tony | 22 |
| 5 | Alice | 22 |
| 6 | Lucy | 23 |
| 7 | Polly | 23 |
| 8 | Tom | 23 |
| 9 | Rose | 24 |
| 10 | Jerry | 24 |
| 11 | Jason | 24 |
| 12 | Lily | 25 |
| 13 | Stephen | 25 |
| 14 | David | 25 |
| 15 | Eddie | 26 |
| 16 | Kate | 27 |
| 17 | Cathy | 27 |
| 18 | Judy | 27 |
| 19 | Julia | 28 |
| 20 | Robert | 28 |
| 21 | Jack | 29 |
+------+---------+-------+
21 rows in set (0.01 sec)
Spark SQL を使用してデータを読み取る
-
Spark ディレクトリで次のコマンドを実行して Spark SQL を開始します。
sh spark-sql -
testデータベースに属するscore_boardテーブルに一時ビューspark_starrocksを作成するために次のコマンドを実行します。spark-sql> CREATE TEMPORARY VIEW spark_starrocks
USING starrocks
OPTIONS
(
"starrocks.table.identifier" = "test.score_board",
"starrocks.fe.http.url" = "<fe_host>:<fe_http_port>",
"starrocks.fe.jdbc.url" = "jdbc:mysql://<fe_host>:<fe_query_port>",
"starrocks.user" = "root",
"starrocks.password" = ""
); -
一時ビューからデータを読み取るために次のコマンドを実行します。
spark-sql> SELECT * FROM spark_starrocks;Spark は次のデータを返します。
1 Bob 21
2 Stan 21
3 Sam 22
4 Tony 22
5 Alice 22
6 Lucy 23
7 Polly 23
8 Tom 23
9 Rose 24
10 Jerry 24
11 Jason 24
12 Lily 25
13 Stephen 25
14 David 25
15 Eddie 26
16 Kate 27
17 Cathy 27
18 Judy 27
19 Julia 28
20 Robert 28
21 Jack 29
Time taken: 1.883 seconds, Fetched 21 row(s)
22/08/09 15:29:36 INFO thriftserver.SparkSQLCLIDriver: Time taken: 1.883 seconds, Fetched 21 row(s)
Spark DataFrame を使用してデータを読み取る
-
Spark ディレクトリで次のコマンドを実行して Spark Shell を開始します。
sh spark-shell -
testデータベースに属するscore_boardテーブルに DataFramestarrocksSparkDFを作成するために次のコマンドを実行します。scala> val starrocksSparkDF = spark.read.format("starrocks")
.option("starrocks.table.identifier", s"test.score_board")
.option("starrocks.fe.http.url", s"<fe_host>:<fe_http_port>")
.option("starrocks.fe.jdbc.url", s"jdbc:mysql://<fe_host>:<fe_query_port>")
.option("starrocks.user", s"root")
.option("starrocks.password", s"")
.load() -
DataFrame からデータを読み取ります。たとえば、最初の10行を読み取りたい場合、次のコマンドを実行します。
scala> starrocksSparkDF.show(10)Spark は次のデータを返します。
+---+-----+-----+
| id| name|score|
+---+-----+-----+
| 1| Bob| 21|
| 2| Stan| 21|
| 3| Sam| 22|
| 4| Tony| 22|
| 5|Alice| 22|
| 6| Lucy| 23|
| 7|Polly| 23|
| 8| Tom| 23|
| 9| Rose| 24|
| 10|Jerry| 24|
+---+-----+-----+
only showing top 10 rows注意
デフォルトでは、読み取りたい行数を指定しない場合、Spark は最初の20行を返します。
Spark RDD を使用してデータを読み取る
-
Spark ディレクトリで次のコマンドを実行して Spark Shell を開始します。
sh spark-shell -
testデータベースに属するscore_boardテーブルに RDDstarrocksSparkRDDを作成するために次のコマンドを実行します。scala> import com.starrocks.connector.spark._
scala> val starrocksSparkRDD = sc.starrocksRDD
(
tableIdentifier = Some("test.score_board"),
cfg = Some(Map(
"starrocks.fenodes" -> "<fe_host>:<fe_http_port>",
"starrocks.request.auth.user" -> "root",
"starrocks.request.auth.password" -> ""
))
) -
RDD からデータを読み取ります。たとえば、最初の10要素を読み取りたい場合、次のコマンドを実行します。
scala> starrocksSparkRDD.take(10)Spark は次のデータを返します。
res0: Array[AnyRef] = Array([1, Bob, 21], [2, Stan, 21], [3, Sam, 22], [4, Tony, 22], [5, Alice, 22], [6, Lucy, 23], [7, Polly, 23], [8, Tom, 23], [9, Rose, 24], [10, Jerry, 24])RDD 全体を読み取るには、次のコマンドを実行します。
scala> starrocksSparkRDD.collect()Spark は次のデータを返します。
res1: Array[AnyRef] = Array([1, Bob, 21], [2, Stan, 21], [3, Sam, 22], [4, Tony, 22], [5, Alice, 22], [6, Lucy, 23], [7, Polly, 23], [8, Tom, 23], [9, Rose, 24], [10, Jerry, 24], [11, Jason, 24], [12, Lily, 25], [13, Stephen, 25], [14, David, 25], [15, Eddie, 26], [16, Kate, 27], [17, Cathy, 27], [18, Judy, 27], [19, Julia, 28], [20, Robert, 28], [21, Jack, 29])
ベストプラクティス
Spark コネクタを使用して StarRocks からデータを読み取る際に、starrocks.filter.query パラメータを使用してフィルタ条件を指定することで、Spark がパーティション、バケット、プレフィックスインデックスをプルーニングし、データ取得のコストを削減することができます。このセクションでは、Spark DataFrame を例にとって、これがどのように達成されるかを示します。
環境設定
| コンポーネント | バージョン |
|---|---|
| Spark | Spark 2.4.4 および Scala 2.11.12 (OpenJDK 64-Bit Server VM, Java 1.8.0_302) |
| StarRocks | 2.2.0 |
| Spark コネクタ | starrocks-spark2_2.11-1.0.0.jar |
データ例
サンプルテーブルを準備するには、次の手順を実行します。
-
testデータベースに移動し、mytableという名前のテーブルを作成します。MySQL [test]> CREATE TABLE `mytable`
(
`k` int(11) NULL COMMENT "bucket",
`b` int(11) NULL COMMENT "",
`dt` datetime NULL COMMENT "",
`v` int(11) NULL COMMENT ""
)
ENGINE=OLAP
DUPLICATE KEY(`k`,`b`, `dt`)
COMMENT "OLAP"
PARTITION BY RANGE(`dt`)
(
PARTITION p202201 VALUES [('2022-01-01 00:00:00'), ('2022-02-01 00:00:00')),
PARTITION p202202 VALUES [('2022-02-01 00:00:00'), ('2022-03-01 00:00:00')),
PARTITION p202203 VALUES [('2022-03-01 00:00:00'), ('2022-04-01 00:00:00'))
)
DISTRIBUTED BY HASH(`k`)
PROPERTIES (
"replication_num" = "3"
); -
mytableにデータを挿入します。MySQL [test]> INSERT INTO mytable
VALUES
(1, 11, '2022-01-02 08:00:00', 111),
(2, 22, '2022-02-02 08:00:00', 222),
(3, 33, '2022-03-02 08:00:00', 333); -
mytableテーブルをクエリします。MySQL [test]> select * from mytable;
+------+------+---------------------+------+
| k | b | dt | v |
+------+------+---------------------+------+
| 1 | 11 | 2022-01-02 08:00:00 | 111 |
| 2 | 22 | 2022-02-02 08:00:00 | 222 |
| 3 | 33 | 2022-03-02 08:00:00 | 333 |
+------+------+---------------------+------+
3 rows in set (0.01 sec)
フルテーブルスキャン
-
Spark ディレクトリで次のコマンドを実行して、
testデータベースに属するmytableテーブルに DataFramedfを作成します。scala> val df = spark.read.format("starrocks")
.option("starrocks.table.identifier", s"test.mytable")
.option("starrocks.fenodes", s"<fe_host>:<fe_http_port>")
.option("user", s"root")
.option("password", s"")
.load() -
StarRocks クラスターの FE ログファイル fe.log を確認し、データを読み取るために実行された SQL ステートメントを見つけます。例：
2022-08-09 18:57:38,091 INFO (nioEventLoopGroup-3-10|196) [TableQueryPlanAction.executeWithoutPassword():126] receive SQL statement [select `k`,`b`,`dt`,`v` from `test`.`mytable`] from external service [ user ['root'@'%']] for database [test] table [mytable] -
testデータベースで、SELECTk,b,dt,vfromtest.mytableステートメントの実行計画を取得するために EXPLAIN を使用します。MySQL [test]> EXPLAIN select `k`,`b`,`dt`,`v` from `test`.`mytable`;
+-----------------------------------------------------------------------+
| Explain String |
+-----------------------------------------------------------------------+
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:1: k | 2: b | 3: dt | 4: v |
| PARTITION: UNPARTITIONED |
| |
| RESULT SINK |
| |
| 1:EXCHANGE |
| |
| PLAN FRAGMENT 1 |
| OUTPUT EXPRS: |
| PARTITION: RANDOM |
| |
| STREAM DATA SINK |
| EXCHANGE ID: 01 |
| UNPARTITIONED |
| |
| 0:OlapScanNode |
| TABLE: mytable |
| PREAGGREGATION: ON |
| partitions=3/3 |
| rollup: mytable |
| tabletRatio=9/9 |
| tabletList=41297,41299,41301,41303,41305,41307,41309,41311,41313 |
| cardinality=3 |
| avgRowSize=4.0 |
| numNodes=0 |
+-----------------------------------------------------------------------+
26 rows in set (0.00 sec)
この例では、プルーニングは行われません。そのため、Spark はデータを保持する3つのパーティションすべて（partitions=3/3 と示されているように）をスキャンし、それらの3つのパーティション内の9つのタブレットすべて（tabletRatio=9/9 と示されているように）をスキャンします。
パーティションプルーニング
-
Spark ディレクトリで次のコマンドを実行し、
starrocks.filter.queryパラメータを使用してパーティションプルーニングのためのフィルタ条件dt='2022-01-02 08:00:00'を指定し、testデータベースに属するmytableテーブルに DataFramedfを作成します。scala> val df = spark.read.format("starrocks")
.option("starrocks.table.identifier", s"test.mytable")
.option("starrocks.fenodes", s"<fe_host>:<fe_http_port>")
.option("user", s"root")
.option("password", s"")
.option("starrocks.filter.query", "dt='2022-01-02 08:00:00'")
.load() -
StarRocks クラスターの FE ログファイル fe.log を確認し、データを読み取るために実行された SQL ステートメントを見つけます。例：
2022-08-09 19:02:31,253 INFO (nioEventLoopGroup-3-14|204) [TableQueryPlanAction.executeWithoutPassword():126] receive SQL statement [select `k`,`b`,`dt`,`v` from `test`.`mytable` where dt='2022-01-02 08:00:00'] from external service [ user ['root'@'%']] for database [test] table [mytable] -
testデータベースで、SELECTk,b,dt,vfromtest.mytablewhere dt='2022-01-02 08:00:00' ステートメントの実行計画を取得するために EXPLAIN を使用します。MySQL [test]> EXPLAIN select `k`,`b`,`dt`,`v` from `test`.`mytable` where dt='2022-01-02 08:00:00';
+------------------------------------------------+
| Explain String |
+------------------------------------------------+
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:1: k | 2: b | 3: dt | 4: v |
| PARTITION: UNPARTITIONED |
| |
| RESULT SINK |
| |
| 1:EXCHANGE |
| |
| PLAN FRAGMENT 1 |
| OUTPUT EXPRS: |
| PARTITION: RANDOM |
| |
| STREAM DATA SINK |
| EXCHANGE ID: 01 |
| UNPARTITIONED |
| |
| 0:OlapScanNode |
| TABLE: mytable |
| PREAGGREGATION: ON |
| PREDICATES: 3: dt = '2022-01-02 08:00:00' |
| partitions=1/3 |
| rollup: mytable |
| tabletRatio=3/3 |
| tabletList=41297,41299,41301 |
| cardinality=1 |
| avgRowSize=20.0 |
| numNodes=0 |
+------------------------------------------------+
27 rows in set (0.01 sec)
この例では、パーティションプルーニングのみが行われ、バケットプルーニングは行われません。そのため、Spark は3つのパーティションのうち1つ（partitions=1/3 と示されているように）をスキャンし、そのパーティション内のすべてのタブレット（tabletRatio=3/3 と示されているように）をスキャンします。
バケットプルーニング
-
Spark ディレクトリで次のコマンドを実行し、
starrocks.filter.queryパラメータを使用してバケットプルーニングのためのフィルタ条件k=1を指定し、testデータベースに属するmytableテーブルに DataFramedfを作成します。scala> val df = spark.read.format("starrocks")
.option("starrocks.table.identifier", s"test.mytable")
.option("starrocks.fenodes", s"<fe_host>:<fe_http_port>")
.option("user", s"root")
.option("password", s"")
.option("starrocks.filter.query", "k=1")
.load() -
StarRocks クラスターの FE ログファイル fe.log を確認し、データを読み取るために実行された SQL ステートメントを見つけます。例：
2022-08-09 19:04:44,479 INFO (nioEventLoopGroup-3-16|208) [TableQueryPlanAction.executeWithoutPassword():126] receive SQL statement [select `k`,`b`,`dt`,`v` from `test`.`mytable` where k=1] from external service [ user ['root'@'%']] for database [test] table [mytable] -
testデータベースで、SELECTk,b,dt,vfromtest.mytablewhere k=1 ステートメントの実行計画を取得するために EXPLAIN を使用します。MySQL [test]> EXPLAIN select `k`,`b`,`dt`,`v` from `test`.`mytable` where k=1;
+------------------------------------------+
| Explain String |
+------------------------------------------+
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:1: k | 2: b | 3: dt | 4: v |
| PARTITION: UNPARTITIONED |
| |
| RESULT SINK |
| |
| 1:EXCHANGE |
| |
| PLAN FRAGMENT 1 |
| OUTPUT EXPRS: |
| PARTITION: RANDOM |
| |
| STREAM DATA SINK |
| EXCHANGE ID: 01 |
| UNPARTITIONED |
| |
| 0:OlapScanNode |
| TABLE: mytable |
| PREAGGREGATION: ON |
| PREDICATES: 1: k = 1 |
| partitions=3/3 |
| rollup: mytable |
| tabletRatio=3/9 |
| tabletList=41299,41305,41311 |
| cardinality=1 |
| avgRowSize=20.0 |
| numNodes=0 |
+------------------------------------------+
27 rows in set (0.01 sec)
この例では、バケットプルーニングのみが行われ、パーティションプルーニングは行われません。そのため、Spark はデータを保持する3つのパーティションすべて（partitions=3/3 と示されているように）をスキャンし、k = 1 フィルタ条件を満たすハッシュ値を取得するために3つのタブレットすべて（tabletRatio=3/9 と示されているように）をスキャンします。
パーティションプルーニングとバケットプルーニング
-
Spark ディレクトリで次のコマンドを実行し、
starrocks.filter.queryパラメータを使用してバケットプルーニングとパーティションプルーニングのための2つのフィルタ条件k=7およびdt='2022-01-02 08:00:00'を指定し、testデータベースに属するmytableテーブルに DataFramedfを作成します。scala> val df = spark.read.format("starrocks")
.option("starrocks.table.identifier", s"test.mytable")
.option("starrocks.fenodes", s"<fe_host>:<fe_http_port>")
.option("user", s"")
.option("password", s"")
.option("starrocks.filter.query", "k=7 and dt='2022-01-02 08:00:00'")
.load() -
StarRocks クラスターの FE ログファイル fe.log を確認し、データを読み取るために実行された SQL ステートメントを見つけます。例：
2022-08-09 19:06:34,939 INFO (nioEventLoopGroup-3-18|212) [TableQueryPlanAction.executeWithoutPassword():126] receive SQL statement [select `k`,`b`,`dt`,`v` from `test`.`mytable` where k=7 and dt='2022-01-02 08:00:00'] from external service [ user ['root'@'%']] for database [test] t
able [mytable] -
testデータベースで、SELECTk,b,dt,vfromtest.mytablewhere k=7 and dt='2022-01-02 08:00:00' ステートメントの実行計画を取得するために EXPLAIN を使用します。MySQL [test]> EXPLAIN select `k`,`b`,`dt`,`v` from `test`.`mytable` where k=7 and dt='2022-01-02 08:00:00';
+----------------------------------------------------------+
| Explain String |
+----------------------------------------------------------+
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:1: k | 2: b | 3: dt | 4: v |
| PARTITION: RANDOM |
| |
| RESULT SINK |
| |
| 0:OlapScanNode |
| TABLE: mytable |
| PREAGGREGATION: ON |
| PREDICATES: 1: k = 7, 3: dt = '2022-01-02 08:00:00' |
| partitions=1/3 |
| rollup: mytable |
| tabletRatio=1/3 |
| tabletList=41301 |
| cardinality=1 |
| avgRowSize=20.0 |
| numNodes=0 |
+----------------------------------------------------------+
17 rows in set (0.00 sec)
この例では、パーティションプルーニングとバケットプルーニングの両方が行われます。そのため、Spark は3つのパーティションのうち1つ（partitions=1/3 と示されているように）をスキャンし、そのパーティション内の1つのタブレット（tabletRatio=1/3 と示されているように）をスキャンします。
プレフィックスインデックスフィルタリング
-
testデータベースに属するmytableテーブルのパーティションにさらにデータレコードを挿入します。MySQL [test]> INSERT INTO mytable
VALUES
(1, 11, "2022-01-02 08:00:00", 111),
(3, 33, "2022-01-02 08:00:00", 333),
(3, 33, "2022-01-02 08:00:00", 333),
(3, 33, "2022-01-02 08:00:00", 333); -
mytableテーブルをクエリします。MySQL [test]> SELECT * FROM mytable;
+------+------+---------------------+------+
| k | b | dt | v |
+------+------+---------------------+------+
| 1 | 11 | 2022-01-02 08:00:00 | 111 |
| 1 | 11 | 2022-01-02 08:00:00 | 111 |
| 3 | 33 | 2022-01-02 08:00:00 | 333 |
| 3 | 33 | 2022-01-02 08:00:00 | 333 |
| 3 | 33 | 2022-01-02 08:00:00 | 333 |
| 2 | 22 | 2022-02-02 08:00:00 | 222 |
| 3 | 33 | 2022-03-02 08:00:00 | 333 |
+------+------+---------------------+------+
7 rows in set (0.01 sec) -
Spark ディレクトリで次のコマンドを実行し、
starrocks.filter.queryパラメータを使用してプレフィックスインデックスフィルタリングのためのフィルタ条件k=1を指定し、testデータベースに属するmytableテーブルに DataFramedfを作成します。scala> val df = spark.read.format("starrocks")
.option("starrocks.table.identifier", s"test.mytable")
.option("starrocks.fenodes", s"<fe_host>:<fe_http_port>")
.option("user", s"root")
.option("password", s"")
.option("starrocks.filter.query", "k=1")
.load() -
testデータベースで、プロファイル報告を有効にするためにis_report_successをtrueに設定します。MySQL [test]> SET is_report_success = true;
Query OK, 0 rows affected (0.00 sec) -
ブラウザを使用して
http://<fe_host>:<http_http_port>/queryページを開き、SELECT * FROM mytable where k=1 ステートメントのプロファイルを確認します。例：OLAP_SCAN (plan_node_id=0):
CommonMetrics:
- CloseTime: 1.255ms
- OperatorTotalTime: 1.404ms
- PeakMemoryUsage: 0.00
- PullChunkNum: 8
- PullRowNum: 2
- __MAX_OF_PullRowNum: 2
- __MIN_OF_PullRowNum: 0
- PullTotalTime: 148.60us
- PushChunkNum: 0
- PushRowNum: 0
- PushTotalTime: 0ns
- SetFinishedTime: 136ns
- SetFinishingTime: 129ns
UniqueMetrics:
- Predicates: 1: k = 1
- Rollup: mytable
- Table: mytable
- BytesRead: 88.00 B
- __MAX_OF_BytesRead: 88.00 B
- __MIN_OF_BytesRead: 0.00
- CachedPagesNum: 0
- CompressedBytesRead: 844.00 B
- __MAX_OF_CompressedBytesRead: 844.00 B
- __MIN_OF_CompressedBytesRead: 0.00
- CreateSegmentIter: 18.582us
- IOTime: 4.425us
- LateMaterialize: 17.385us
- PushdownPredicates: 3
- RawRowsRead: 2
- __MAX_OF_RawRowsRead: 2
- __MIN_OF_RawRowsRead: 0
- ReadPagesNum: 12
- __MAX_OF_ReadPagesNum: 12
- __MIN_OF_ReadPagesNum: 0
- RowsRead: 2
- __MAX_OF_RowsRead: 2
- __MIN_OF_RowsRead: 0
- ScanTime: 154.367us
- SegmentInit: 95.903us
- BitmapIndexFilter: 0ns
- BitmapIndexFilterRows: 0
- BloomFilterFilterRows: 0
- ShortKeyFilterRows: 3
- __MAX_OF_ShortKeyFilterRows: 3
- __MIN_OF_ShortKeyFilterRows: 0
- ZoneMapIndexFilterRows: 0
- SegmentRead: 2.559us
- BlockFetch: 2.187us
- BlockFetchCount: 2
- __MAX_OF_BlockFetchCount: 2
- __MIN_OF_BlockFetchCount: 0
- BlockSeek: 7.789us
- BlockSeekCount: 2
- __MAX_OF_BlockSeekCount: 2
- __MIN_OF_BlockSeekCount: 0
- ChunkCopy: 25ns
- DecompressT: 0ns
- DelVecFilterRows: 0
- IndexLoad: 0ns
- PredFilter: 353ns
- PredFilterRows: 0
- RowsetsReadCount: 7
- SegmentsReadCount: 3
- __MAX_OF_SegmentsReadCount: 2
- __MIN_OF_SegmentsReadCount: 0
- TotalColumnsDataPageCount: 8
- __MAX_OF_TotalColumnsDataPageCount: 8
- __MIN_OF_TotalColumnsDataPageCount: 0
- UncompressedBytesRead: 508.00 B
- __MAX_OF_UncompressedBytesRead: 508.00 B
- __MIN_OF_UncompressedBytesRead: 0.00
この例では、フィルタ条件 k = 1 がプレフィックスインデックスにヒットすることができます。そのため、Spark は3行をフィルタリングすることができます（ShortKeyFilterRows: 3 と示されているように）。