Spark コネクタを使用して StarRocks からデータを読み取る
StarRocks は、Apache Spark™ 用に独自開発したコネクタである StarRocks Connector for Apache Spark™（以下、Spark コネクタ）を提供しています。これを使用して、Spark を通じて StarRocks テーブルからデータを読み取ることができます。Spark を使用して、StarRocks から読み取ったデータに対して複雑な処理や機械学習を行うことができます。
Spark コネクタは、Spark SQL、Spark DataFrame、Spark RDD の3つの読み取り方法をサポートしています。
Spark SQL を使用して StarRocks テーブルに一時ビューを作成し、その一時ビューを通じて直接 StarRocks テーブルからデータを読み取ることができます。
また、StarRocks テーブルを 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 tablets の数。このパラメータの値が小さいほど、生成される Spark RDD パーティションの数が多くなります。Spark RDD パーティションの数が多いほど、Spark の並行性は高くなりますが、StarRocks への負荷も大きくなります。 |
| starrocks.batch.size | 4096 | 一度に BEs から読み取ることができる最大行数。このパラメータの値を増やすことで、Spark と StarRocks 間で確立される接続の数を減らし、ネットワーク遅延による余分な時間オーバーヘッドを軽減できます。 |
| starrocks.exec.mem.limit | 2147483648 | クエリごとに許可される最大メモリ量。単位: バイト。デフォルトのメモリ制限は2GBです。 |
| 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 コネクタは JDBC を使用して StarRocks にアクセスし、より詳細なテーブル情報を取得します。そのため、
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 コネクタ 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つの tablets すべて（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 と示されているように）をスキャンし、そのパーティション内のすべての tablets（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 フィルタ条件を満たす Hash 値を取得するために3つの tablets（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つの tablet（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 と示されているように）をフィルタリングすることができます。