2026年5月17日に最終更新

Spark Doris Connector

Spark Doris Connector は、Apache Doris と Apache Spark を接続するコネクタです。Spark から Doris に保存されたデータを読み取り、Spark から Doris にデータを書き込めます。リポジトリ: apache/doris-spark-connector。

主な機能は次のとおりです。

ユースケース	推奨する方法	説明
Doris データのバッチ読み取り	DataFrame、Spark SQL	RDD にも対応しています。DataFrame または Spark SQL を推奨します。
Doris データのバッチ書き込み	DataFrame、Spark SQL	書き込む列を指定できます。バージョン 1.3.0 以降は Overwrite モードにも対応しています。
Doris データのストリーミング書き込み	Structured Streaming	通常の構造化データの書き込みに加え、DataFrame の最初の列をそのまま書き込めます。
Doris データの高速読み取り	Arrow Flight SQL	バージョン 24.0.0 以降で対応しています。Doris バージョン 2.1.0 以降が必要です。
Catalog 経由での Doris へのアクセス	Spark Doris Catalog	バージョン 24.0.0 以降で対応しています。Spark Catalog から Doris のデータベースとテーブルを管理できます。

はじめる前に

バージョンの互換性

まず、使用する Spark、Doris、Java、Scala のバージョンに対応する Connector のバージョンを選択します。

Connector	Spark	Doris	Java	Scala
26.0.0	3.5 - 3.1, 2.4	1.0 +	8	2.12, 2.11
25.2.0	3.5 - 3.1, 2.4	1.0 +	8	2.12, 2.11
25.1.0	3.5 - 3.1, 2.4	1.0 +	8	2.12, 2.11
25.0.1	3.5 - 3.1, 2.4	1.0 +	8	2.12, 2.11
25.0.0	3.5 - 3.1, 2.4	1.0 +	8	2.12, 2.11
1.3.2	3.4 - 3.1, 2.4, 2.3	1.0 - 2.1.6	8	2.12, 2.11
1.3.1	3.4 - 3.1, 2.4, 2.3	1.0 - 2.1.0	8	2.12, 2.11
1.3.0	3.4 - 3.1, 2.4, 2.3	1.0 - 2.1.0	8	2.12, 2.11
1.2.0	3.2, 3.1, 2.3	1.0 - 2.0.2	8	2.12, 2.11
1.1.0	3.2, 3.1, 2.3	1.0 - 1.2.8	8	2.12, 2.11
1.0.1	3.1, 2.3	0.12 - 0.15	8	2.12, 2.11

Maven で依存関係を追加する

プロジェクトの pom.xml に Spark Doris Connector の依存関係を追加します。使用する Spark と Connector のバージョンに応じて、artifactId と version を変更してください。

<dependency>
    <groupId>org.apache.doris</groupId>
    <artifactId>spark-doris-connector-spark-3.5</artifactId>
    <version>25.2.0</version>
</dependency>

ヒント

バージョン 24.0.0 以降、Doris Connector のパッケージ命名規則が変更されています。

Scala のバージョン情報はパッケージ名に含まれなくなりました。
Spark 2.x では、共通のパッケージ名 spark-doris-connector-spark-2 を使用します。デフォルトでは Scala 2.11 向けにのみコンパイルされます。Scala 2.12 版が必要な場合は、自分でコンパイルしてください。
Spark 3.x では、使用する Spark のバージョンに対応する spark-doris-connector-spark-3.x パッケージを使用します。Spark 3.0 では spark-doris-connector-spark-3.1 パッケージを使用できます。

対応するバージョンの Jar ファイルは Maven リポジトリからダウンロードすることもできます。

ソースからコンパイルする

ソースコードを自分でコンパイルする場合は、ソースディレクトリで sh build.sh を実行し、プロンプトに従って必要な Scala と Spark のバージョンを入力します。

コンパイルに成功すると、dist ディレクトリに対象の Jar ファイル（例: spark-doris-connector-spark-3.5-25.2.0.jar）が生成されます。このファイルを Spark の classpath に配置すると、Spark Doris Connector を使用できます。

Spark の実行モード	Jar ファイルの配置方法
ローカルモード	Jar ファイルを `jars/` ディレクトリに配置します。
Yarn クラスターモード	Jar ファイルを事前にデプロイするパッケージに配置します。

たとえば、spark-doris-connector-spark-3.5-25.2.0.jar を HDFS にアップロードし、spark.yarn.jars で依存関係を追加します。

# 1. Upload spark-doris-connector-spark-3.5-25.2.0.jar to HDFS
hdfs dfs -mkdir /spark-jars/
hdfs dfs -put /your_local_path/spark-doris-connector-spark-3.5-25.2.0.jar /spark-jars/

# 2. Add the spark-doris-connector-spark-3.5-25.2.0.jar dependency in the cluster
spark.yarn.jars=hdfs:///spark-jars/spark-doris-connector-spark-3.5-25.2.0.jar

シナリオ 1: Doris データをバッチで読み取る

Spark Doris Connector は、DataFrame、Spark SQL、RDD、PySpark による Doris データの読み取りに対応しています。Doris データの読み取りには DataFrame または Spark SQL を推奨します。

DataFrame で読み取る

val dorisSparkDF = spark.read.format("doris")
    .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
    .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
    .option("user", "$YOUR_DORIS_USERNAME")
    .option("password", "$YOUR_DORIS_PASSWORD")
    .load()

dorisSparkDF.show(5)

Spark SQL で読み取る

CREATE TEMPORARY VIEW spark_doris
USING doris
OPTIONS(
    "table.identifier"="$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME",
    "fenodes"="$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT",
    "user"="$YOUR_DORIS_USERNAME",
    "password"="$YOUR_DORIS_PASSWORD"
);

SELECT * FROM spark_doris;

RDD で読み取る

import org.apache.doris.spark._

val dorisSparkRDD = sc.dorisRDD(
    tableIdentifier = Some("$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME"),
    cfg = Some(Map(
        "doris.fenodes" -> "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT",
        "doris.request.auth.user" -> "$YOUR_DORIS_USERNAME",
        "doris.request.auth.password" -> "$YOUR_DORIS_PASSWORD"
    ))
)

dorisSparkRDD.collect()

PySpark で読み取る

dorisSparkDF = spark.read.format("doris")
    .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
    .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
    .option("user", "$YOUR_DORIS_USERNAME")
    .option("password", "$YOUR_DORIS_PASSWORD")
    .load()

# Show 5 rows of data
dorisSparkDF.show(5)

Arrow Flight SQL で高速に読み取る

バージョン 24.0.0 以降、Spark Doris Connector は Arrow Flight SQL によるデータ読み取りに対応しています。この方法には Doris バージョン 2.1.0 以降が必要です。

次のパラメータを設定します。

パラメータ	説明
`doris.read.mode`	`arrow` に設定すると、Arrow Flight SQL でデータを読み取ります。
`doris.read.arrow-flight-sql.port`	FE に設定された Arrow Flight SQL ポートを指定します。

サーバー側の設定については、Arrow Flight SQL に基づく高速データ転送リンクを参照してください。

val df = spark.read.format("doris")
    .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
    .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
    .option("doris.user", "$YOUR_DORIS_USERNAME")
    .option("doris.password", "$YOUR_DORIS_PASSWORD")
    .option("doris.read.mode", "arrow")
    .option("doris.read.arrow-flight-sql.port", "12345")
    .load()

df.show()

シナリオ 2: Doris データをバッチで書き込む

Spark Doris Connector は、DataFrame と Spark SQL による Doris データのバッチ書き込みに対応しています。

DataFrame で書き込む

val mockDataDF = List(
    (3, "440403001005", "21.cn"),
    (1, "4404030013005", "22.cn"),
    (33, null, "23.cn")
).toDF("id", "mi_code", "mi_name")

mockDataDF.show(5)

mockDataDF.write.format("doris")
    .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
    .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
    .option("user", "$YOUR_DORIS_USERNAME")
    .option("password", "$YOUR_DORIS_PASSWORD")
    // Specify the columns to write
    .option("doris.write.fields", "$YOUR_FIELDS_TO_WRITE")
    // Overwrite is supported starting from version 1.3.0
    // .mode(SaveMode.Overwrite)
    .save()

Spark SQL で書き込む

CREATE TEMPORARY VIEW spark_doris
USING doris
OPTIONS(
    "table.identifier"="$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME",
    "fenodes"="$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT",
    "user"="$YOUR_DORIS_USERNAME",
    "password"="$YOUR_DORIS_PASSWORD"
);

INSERT INTO spark_doris VALUES ("VALUE1", "VALUE2", ...);

-- insert into select
INSERT INTO spark_doris SELECT * FROM YOUR_TABLE;

-- insert overwrite
INSERT OVERWRITE spark_doris SELECT * FROM YOUR_TABLE;

シナリオ 3: Doris データをストリーミングで書き込む

Spark Doris Connector は、Structured Streaming から Doris への書き込みに対応しています。データが Doris テーブルのスキーマに適合しているかどうかに応じて、構造化データとして書き込む方法と、最初の列をそのまま書き込む方法を選べます。

構造化データを書き込む

val df = spark.readStream.format("your_own_stream_source").load()

df.writeStream
    .format("doris")
    .option("checkpointLocation", "$YOUR_CHECKPOINT_LOCATION")
    .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
    .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
    .option("user", "$YOUR_DORIS_USERNAME")
    .option("password", "$YOUR_DORIS_PASSWORD")
    .start()
    .awaitTermination()

データストリームの最初の列をそのまま書き込む

データストリームの最初の列が Doris テーブルのスキーマに適合している場合、たとえば列順が同じ CSV データや、フィールド名が一致する JSON データの場合は、doris.sink.streaming.passthrough を true に設定します。これにより、内容を複数の DataFrame 列に展開せず、最初の列をそのまま書き込めます。

次の例では Kafka をソースとして使用します。書き込み先の Doris テーブルに次のスキーマがあるとします。

CREATE TABLE `t2` (
    `c0` int NULL,
    `c1` varchar(10) NULL,
    `c2` date NULL
) ENGINE=OLAP
DUPLICATE KEY(`c0`)
COMMENT 'OLAP'
DISTRIBUTED BY HASH(`c0`) BUCKETS 1
PROPERTIES (
    "replication_allocation" = "tag.location.default: 1"
);

Kafka メッセージの value は、{"c0":1,"c1":"a","c2":"2024-01-01"} 形式の JSON データです。

val kafkaSource = spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "$YOUR_KAFKA_SERVERS")
    .option("startingOffsets", "latest")
    .option("subscribe", "$YOUR_KAFKA_TOPICS")
    .load()

// Select value as the first column of the DataFrame
kafkaSource.selectExpr("CAST(value as STRING)")
    .writeStream
    .format("doris")
    .option("checkpointLocation", "$YOUR_CHECKPOINT_LOCATION")
    .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
    .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
    .option("user", "$YOUR_DORIS_USERNAME")
    .option("password", "$YOUR_DORIS_PASSWORD")
    // Setting this option to true writes the first column of the DataFrame directly
    .option("doris.sink.streaming.passthrough", "true")
    .option("doris.sink.properties.format", "json")
    .start()
    .awaitTermination()

JSON 形式で書き込む

doris.sink.properties.format を json に設定すると、Connector はデータを JSON 形式で Doris に書き込みます。

val df = spark.readStream.format("your_own_stream_source").load()

df.writeStream
    .format("doris")
    .option("checkpointLocation", "$YOUR_CHECKPOINT_LOCATION")
    .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
    .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
    .option("user", "$YOUR_DORIS_USERNAME")
    .option("password", "$YOUR_DORIS_PASSWORD")
    .option("doris.sink.properties.format", "json")
    .start()
    .awaitTermination()

シナリオ 4: Spark Doris Catalog 経由で Doris にアクセスする

バージョン 24.0.0 以降、Spark Doris Connector は Spark Catalog 経由での Doris へのアクセスに対応しています。

Catalog の設定項目

オプション名	必須	説明
`spark.sql.catalog.your_catalog_name`	はい	Catalog プロバイダーのクラス名を指定します。Doris で有効な値は `org.apache.doris.spark.catalog.DorisTableCatalog` のみです。
`spark.sql.catalog.your_catalog_name.doris.fenodes`	はい	Doris FE ノードを `fe_ip:fe_http_port` 形式で指定します。
`spark.sql.catalog.your_catalog_name.doris.query.port`	いいえ	Doris FE のクエリポートを指定します。`spark.sql.catalog.your_catalog_name.doris.fe.auto.fetch` が `true` の場合は省略できます。
`spark.sql.catalog.your_catalog_name.doris.user`	はい	Doris のユーザーを指定します。
`spark.sql.catalog.your_catalog_name.doris.password`	はい	Doris のパスワードを指定します。
`spark.sql.defaultCatalog`	いいえ	Spark SQL のデフォルト Catalog を指定します。

ヒント

DataFrame と Spark SQL に適用できる Connector のパラメータは、すべて Catalog にも設定できます。たとえば、JSON 形式でデータを書き込むには、spark.sql.catalog.your_catalog_name.doris.sink.properties.format を json に設定します。

DataFrame プログラムで Catalog を使用する

val conf = new SparkConf()
conf.set("spark.sql.catalog.your_catalog_name", "org.apache.doris.spark.catalog.DorisTableCatalog")
conf.set("spark.sql.catalog.your_catalog_name.doris.fenodes", "192.168.0.1:8030")
conf.set("spark.sql.catalog.your_catalog_name.doris.query.port", "9030")
conf.set("spark.sql.catalog.your_catalog_name.doris.user", "root")
conf.set("spark.sql.catalog.your_catalog_name.doris.password", "")
val spark = builder.config(conf).getOrCreate()
spark.sessionState.catalogManager.setCurrentCatalog("your_catalog_name")

// show all databases
spark.sql("show databases")

// use databases
spark.sql("use your_doris_db")

// show tables in test
spark.sql("show tables")

// query table
spark.sql("select * from your_doris_table")

// write data
spark.sql("insert into your_doris_table values(xxx)")

Spark SQL CLI で Catalog を使用する

必要なパラメータを設定して Spark SQL CLI を起動します。

spark-sql \
--conf "spark.sql.catalog.your_catalog_name=org.apache.doris.spark.catalog.DorisTableCatalog" \
--conf "spark.sql.catalog.your_catalog_name.doris.fenodes=192.168.0.1:8030" \
--conf "spark.sql.catalog.your_catalog_name.doris.query.port=9030" \
--conf "spark.sql.catalog.your_catalog_name.doris.user=root" \
--conf "spark.sql.catalog.your_catalog_name.doris.password=" \
--conf "spark.sql.defaultCatalog=your_catalog_name"

Spark SQL CLI でクエリを実行します。

-- show all databases
show databases;

-- use databases
use your_doris_db;

-- show tables in test
show tables;

-- query table
select * from your_doris_table;

-- write data
insert into your_doris_table values(xxx);
insert into your_doris_table select * from your_source_table;

-- access table with full name
select * from your_catalog_name.your_doris_db.your_doris_table;
insert into your_catalog_name.your_doris_db.your_doris_table values(xxx);
insert into your_catalog_name.your_doris_db.your_doris_table select * from your_source_table;

Java の例

Java 版の例は samples/doris-demo/spark-demo/ にあります。apache/doris のサンプルディレクトリを参照してください。

設定リファレンス

共通の設定項目

キー	デフォルト値	説明
`doris.fenodes`	--	Doris FE の HTTP アドレス。複数指定する場合はカンマで区切ります。
`doris.table.identifier`	--	Doris テーブル名。例: `db1.tbl1`。
`doris.user`	--	Doris にアクセスするユーザー名。
`doris.password`	空文字列	Doris にアクセスするためのパスワード。
`doris.request.retries`	3	Doris へのリクエスト送信時の再試行回数。
`doris.request.connect.timeout.ms`	30000	Doris へのリクエスト送信時の接続タイムアウト。
`doris.request.read.timeout.ms`	30000	Doris へのリクエスト送信時の読み取りタイムアウト。
`doris.request.query.timeout.s`	21600	Doris クエリのタイムアウト。デフォルトは 6 時間です。`-1` はタイムアウトなしを意味します。
`doris.request.tablet.size`	1	一つの RDD Partition に対応する Doris Tablet の数。値を小さくすると Partition の数が増え、Spark 側の並列度が上がりますが、Doris への負荷も増えます。
`doris.read.field`	--	Doris テーブルから読み取る列名のリスト。カンマで区切ります。
`doris.batch.size`	4064	一回のバッチで BE から読み取る行数の上限。値を大きくすると Spark と Doris 間で確立する接続数が減り、ネットワーク遅延による追加の時間的負荷が軽減されます。
`doris.exec.mem.limit`	8589934592	一つのクエリのメモリ上限。デフォルトは 8 GB で、単位はバイトです。
`doris.write.fields`	--	Doris テーブルに書き込むフィールド、またはフィールドの順序をカンマ区切りで指定します。デフォルトでは、Doris テーブルのフィールド順にすべてのフィールドを書き込みます。
`doris.sink.batch.size`	500000	一回のバッチで BE に書き込む行数の上限。
`doris.sink.max-retries`	0	BE への書き込みが失敗した後の再試行回数。バージョン 1.3.0 以降のデフォルトは 0 で、再試行しません。この値が 0 より大きい場合はバッチ単位で再試行し、`doris.sink.batch.size` 件分のデータが Spark Executor のメモリにキャッシュされます。そのため、必要に応じてメモリ割り当てを増やしてください。
`doris.sink.retry.interval.ms`	10000	再試行回数を設定した場合の、再試行間隔（ミリ秒）。
`doris.sink.properties.format`	csv	Stream Load のデータ形式。`csv`、`json`、`arrow` の三形式に対応します。その他のパラメータはStream Load マニュアルを参照してください。
`doris.sink.properties.*`	--	Stream Load のインポートパラメータ。たとえば、`'doris.sink.properties.column_separator' = ','` で列区切り文字を指定します。その他のパラメータはStream Load マニュアルを参照してください。
`doris.sink.task.partition.size`	--	Doris への書き込みタスクに対応する Partition 数。フィルタリングなどの Spark RDD 操作後は、書き込む Partition 数が多くても、Partition ごとのレコード数は少ない場合があります。その場合、書き込み頻度が上がり、計算リソースが無駄になります。この値を小さくすると Doris への書き込み頻度と compaction の負荷が下がります。`doris.sink.task.use.repartition` と併用します。
`doris.sink.task.use.repartition`	false	Doris に書き込む Partition 数の制御に repartition を使用するかどうか。デフォルトの `false` では coalesce を使用します。書き込み前に Spark の action 演算子がない場合、全体の計算並列度が下がる可能性があります。`true` では repartition を使用します。最終的な Partition 数を設定できますが、追加の shuffle 負荷が発生します。
`doris.sink.batch.interval.ms`	0	sink バッチ間の間隔（ミリ秒）。
`doris.sink.enable-2pc`	false	二相コミットを有効にするかどうか。有効にすると、ジョブの終了時にトランザクションがコミットされ、一部のタスクが失敗した場合は pre-commit 状態のすべてのトランザクションがロールバックされます。
`doris.sink.auto-redirect`	true	Stream Load リクエストをリダイレクトするかどうか。有効にすると Stream Load は FE 経由で書き込まれ、BE の情報を明示的に取得しなくなります。
`doris.enable.https`	false	FE への HTTPS リクエストを有効にするかどうか。
`doris.https.key-store-path`	-	HTTPS キーストアのパス。
`doris.https.key-store-type`	JKS	HTTPS キーストアの種類。
`doris.https.key-store-password`	-	HTTPS キーストアのパスワード。
`doris.read.mode`	thrift	Doris の読み取りモード。指定できる値は `thrift` と `arrow` です。
`doris.read.arrow-flight-sql.port`	-	Doris FE の Arrow Flight SQL ポート。`doris.read.mode` が `arrow` の場合、Arrow Flight SQL によるデータ読み取りに使用します。サーバー側の設定については、Arrow Flight SQL に基づく高速データ転送リンクを参照してください。
`doris.sink.label.prefix`	spark-doris	Stream Load で書き込む際のインポートラベルのプレフィックス。
`doris.thrift.max.message.size`	2147483647	Thrift によるデータ読み取り時のメッセージサイズの上限。
`doris.fe.auto.fetch`	false	FE 情報を自動取得するかどうか。`true` に設定すると、`doris.fenodes` に設定したノードを基にすべての FE ノード情報を取得するため、複数のノードや `doris.read.arrow-flight-sql.port` と `doris.query.port` を個別に設定する必要はありません。
`doris.read.bitmap-to-string`	false	読み取り時に Bitmap 型を配列インデックスからなる文字列に変換するかどうか。結果の形式については、関数定義の BITMAP_TO_STRING を参照してください。
`doris.read.bitmap-to-base64`	false	読み取り時に Bitmap 型を Base64 エンコードされた文字列に変換するかどうか。結果の形式については、関数定義の BITMAP_TO_BASE64 を参照してください。
`doris.query.port`	-	Doris FE のクエリポート。上書き書き込みと Catalog のメタデータ取得に使用します。

SQL と DataFrame 固有の設定

キー	デフォルト値	説明
`doris.filter.query.in.max.count`	10000	述語プッシュダウンで `in` 式に指定できる値の最大数。この数を超えると、`in` 式のフィルターは Spark 側で処理されます。

Structured Streaming 固有の設定

キー	デフォルト値	説明
`doris.sink.streaming.passthrough`	false	最初の列の値を処理せずにそのまま書き込みます。

RDD 固有の設定

キー	デフォルト値	説明
`doris.request.auth.user`	--	Doris にアクセスするユーザー名。
`doris.request.auth.password`	--	Doris にアクセスするためのパスワード。
`doris.filter.query`	--	読み取るデータを絞り込む式。この式は Doris にそのまま渡され、Doris がソースデータのフィルタリングに使用します。

型の対応関係

Doris から Spark への列の型の対応関係

Doris の型	Spark の型
NULL_TYPE	DataTypes.NullType
BOOLEAN	DataTypes.BooleanType
TINYINT	DataTypes.ByteType
SMALLINT	DataTypes.ShortType
INT	DataTypes.IntegerType
BIGINT	DataTypes.LongType
FLOAT	DataTypes.FloatType
DOUBLE	DataTypes.DoubleType
DATE	DataTypes.DateType
DATETIME	DataTypes.TimestampType
DECIMAL	DecimalType
CHAR	DataTypes.StringType
LARGEINT	DecimalType
VARCHAR	DataTypes.StringType
STRING	DataTypes.StringType
JSON	DataTypes.StringType
VARIANT	DataTypes.StringType
TIME	DataTypes.DoubleType
HLL	DataTypes.StringType
Bitmap	DataTypes.StringType

Spark から Doris へのデータ型の対応関係

Spark の型	Doris の型
BooleanType	BOOLEAN
ShortType	SMALLINT
IntegerType	INT
LongType	BIGINT
FloatType	FLOAT
DoubleType	DOUBLE
DecimalType	DECIMAL
StringType	VARCHAR/STRING
DateType	DATE
TimestampType	DATETIME
ArrayType	ARRAY
MapType	MAP/JSON
StructType	STRUCT/JSON

ヒント

バージョン 24.0.0 以降、Bitmap 型を読み取ったときの戻り値の型は string で、デフォルトで返される文字列は Read unsupported です。

FAQ とトラブルシューティング

Bitmap 型を書き込むには？

Spark SQL で INSERT INTO を使って書き込むとき、書き込み先の Doris テーブルに BITMAP 型または HLL 型のデータが含まれる場合は、doris.ignore-type を該当する型に設定し、doris.write.fields で列の対応付けと変換を指定します。

BITMAP

CREATE TEMPORARY VIEW spark_doris
USING doris
OPTIONS(
    "table.identifier"="$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME",
    "fenodes"="$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT",
    "user"="$YOUR_DORIS_USERNAME",
    "password"="$YOUR_DORIS_PASSWORD",
    "doris.ignore-type"="bitmap",
    "doris.write.fields"="col1,col2,col3,bitmap_col2=to_bitmap(col2),bitmap_col3=bitmap_hash(col3)"
);

HLL

CREATE TEMPORARY VIEW spark_doris
USING doris
OPTIONS(
    "table.identifier"="$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME",
    "fenodes"="$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT",
    "user"="$YOUR_DORIS_USERNAME",
    "password"="$YOUR_DORIS_PASSWORD",
    "doris.ignore-type"="hll",
    "doris.write.fields"="col1,hll_col1=hll_hash(col1)"
);

ヒント

バージョン 24.0.0 以降、doris.ignore-type は非推奨です。書き込み時にこのパラメータを追加する必要はありません。

Overwrite 書き込みを使用するには？

バージョン 1.3.0 以降、Connector は書き込み時の Overwrite モードに対応しています。Overwrite で置き換えられるのはテーブル全体のデータのみです。使用方法は次のとおりです。

DataFrame

resultDf.write.format("doris")
    .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
    // your own options
    .mode(SaveMode.Overwrite)
    .save()

SQL

INSERT OVERWRITE your_target_table SELECT * FROM your_source_table;

Bitmap 型を読み取るには？

バージョン 24.0.0 以降、Connector は Arrow Flight SQL を使った変換済み Bitmap データの読み取りに対応しています。この機能には Doris バージョン 2.1.0 以降が必要です。

Bitmap から String への変換

DataFrame の場合、doris.read.bitmap-to-string を true に設定します。結果の形式については、オプションの定義を参照してください。

spark.read.format("doris")
    .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
    .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
    .option("user", "$YOUR_DORIS_USERNAME")
    .option("password", "$YOUR_DORIS_PASSWORD")
    .option("doris.read.bitmap-to-string", "true")
    .load()

Bitmap から Base64 への変換

DataFrame の場合、doris.read.bitmap-to-base64 を true に設定します。結果の形式については、オプションの定義を参照してください。

spark.read.format("doris")
    .option("doris.table.identifier", "$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
    .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
    .option("user", "$YOUR_DORIS_USERNAME")
    .option("password", "$YOUR_DORIS_PASSWORD")
    .option("doris.read.bitmap-to-base64", "true")
    .load()

DataFrame の書き込み中に ErrorIfExists エラーが発生した場合は？

DataFrame の書き込みが org.apache.spark.sql.AnalysisException: TableProvider implementation doris cannot be written with ErrorIfExists mode, please use Append or Overwrite modes instead. というエラーで失敗した場合は、保存モードを Append に設定します。

resultDf.write.format("doris")
    .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
    // your own options
    .mode(SaveMode.Append)
    .save()

はじめる前に​

バージョンの互換性​

Maven で依存関係を追加する​

ソースからコンパイルする​

シナリオ 1: Doris データをバッチで読み取る​

DataFrame で読み取る​

Spark SQL で読み取る​

RDD で読み取る​

PySpark で読み取る​

Arrow Flight SQL で高速に読み取る​

シナリオ 2: Doris データをバッチで書き込む​

DataFrame で書き込む​

Spark SQL で書き込む​

シナリオ 3: Doris データをストリーミングで書き込む​

構造化データを書き込む​

データストリームの最初の列をそのまま書き込む​

JSON 形式で書き込む​

シナリオ 4: Spark Doris Catalog 経由で Doris にアクセスする​

Catalog の設定項目​

DataFrame プログラムで Catalog を使用する​

Spark SQL CLI で Catalog を使用する​

Java の例​

設定リファレンス​

共通の設定項目​

SQL と DataFrame 固有の設定​

Structured Streaming 固有の設定​

RDD 固有の設定​

型の対応関係​

Doris から Spark への列の型の対応関係​

Spark から Doris へのデータ型の対応関係​

FAQ とトラブルシューティング​

Bitmap 型を書き込むには？​

Overwrite 書き込みを使用するには？​

Bitmap 型を読み取るには？​

DataFrame の書き込み中に ErrorIfExists エラーが発生した場合は？​