バージョン: 4.x

Spark Doris Connector

Spark Doris Connectorは、Sparkを通じてDorisに格納されたデータの読み取りと、Sparkを通じてDorisへのデータの書き込みをサポートします。

コードリポジトリ: https://github.com/apache/doris-spark-connector

RDD、DataFrame、Spark SQLメソッドを通じてDorisからのバッチデータ読み取りをサポートします。DataFrameまたはSpark SQLの使用を推奨します。
DataFrameとSpark SQLを使用したDorisへのバッチまたはストリーミングデータ書き込みをサポートします。
データ転送量を削減するためにDoris側でのデータフィルタリングをサポートします。

バージョン互換性

Connector	Spark	Doris	Java	Scala
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

<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パッケージを使用できます。

注意事項

異なるSparkとScalaバージョンに応じて、対応するConnectorバージョンを置き換えてください。
関連するバージョンのjarパッケージはこちらからダウンロードすることもできます。

コンパイル

ソースコードディレクトリでsh build.shを実行し、プロンプトに従ってコンパイルに必要なScalaとSparkバージョンを入力してください。

コンパイルが成功すると、distディレクトリにターゲットjarパッケージが生成されます。例：spark-doris-connector-spark-3.5-25.2.0.jar。このファイルをSparkのClassPathにコピーしてSpark-Doris-Connectorを使用してください。

例えば、SparkをLocalモードで実行する場合は、このファイルをjars/フォルダに配置してください。SparkをYarnクラスターモードで実行する場合は、このファイルを事前デプロイメントパッケージに配置してください。

例えば、spark-doris-connector-spark-3.5-25.2.0.jarをHDFSにアップロードし、HDFS上のjarパッケージパスを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 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

使用例

バッチ読み取り

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_RESFUL_PORT",
        "doris.request.auth.user" -> "$YOUR_DORIS_USERNAME",
        "doris.request.auth.password" -> "$YOUR_DORIS_PASSWORD"
    ))
)

dorisSparkRDD.collect()

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_RESFUL_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_RESFUL_PORT",
    "user"="$YOUR_DORIS_USERNAME",
    "password"="$YOUR_DORIS_PASSWORD"
);

SELECT * FROM spark_doris;

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_RESFUL_PORT")
        .option("user", "$YOUR_DORIS_USERNAME")
        .option("password", "$YOUR_DORIS_PASSWORD")
        .load()
# show 5 lines data 
dorisSparkDF.show(5)

Arrow Flight SQLによるデータ読み取り

バージョン24.0.0以降、Arrow Flight SQLによるデータ読み取りがサポートされています（Dorisバージョン >= 2.1.0が必要）。

doris.read.modeをarrowに設定し、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_RESFUL_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()

バッチ書き込み

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_RESFUL_PORT")
        .option("user", "$YOUR_DORIS_USERNAME")
        .option("password", "$YOUR_DORIS_PASSWORD")
        // Other options
        // Specify columns to write
        .option("doris.write.fields", "$YOUR_FIELDS_TO_WRITE")
        // Starting from version 1.3.0, overwrite write is supported
        // .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_RESFUL_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 SELECT * FROM YOUR_TABLE;

ストリーミング書き込み

DataFrame

構造化データ書き込み

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_RESFUL_PORT")
        .option("user", "$YOUR_DORIS_USERNAME")
        .option("password", "$YOUR_DORIS_PASSWORD")
        .start()
        .awaitTermination()

Direct Write

データストリームの最初の列がDorisテーブル構造に準拠している場合、例えば同じ列順序のCSVデータや一貫したフィールド名を持つJSONデータなどの場合、doris.sink.streaming.passthroughオプションをtrueに設定することで、この列のデータをDataFrameに変換することなく直接書き込むことができます。

Kafkaソースを例に取ると：

書き込み対象のテーブル構造は以下の通りであると仮定します：

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"
);

メッセージの値はJSON形式です：{"c0":1,"c1":"a","dt":"2024-01-01"}。

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 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_RESFUL_PORT")
        .option("user", "$YOUR_DORIS_USERNAME")
        .option("password", "$YOUR_DORIS_PASSWORD")
        // Setting this option to true will directly write the first column of DataFrame
        .option("doris.sink.streaming.passthrough", "true")
        .option("doris.sink.properties.format", "json")
        .start()
        .awaitTermination()

JSON形式での書き込み

doris.sink.properties.formatをjsonに設定します。

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

df.write.format("doris")
        .option("doris.fenodes", "$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_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")
        .save()

Spark Doris Catalog

バージョン24.0.0以降、Spark Catalogを通じたDorisへのアクセスがサポートされています。

Catalog Config

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

ヒント

DataFrameとSpark SQLに適用可能なすべてのコネクタパラメータをcatalogに設定できます。
例えば、データをjson形式で書き込みたい場合は、オプションspark.sql.catalog.your_catalog_name.doris.sink.properties.formatをjsonに設定できます。

DataFrame

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

必要なパラメータを設定し、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 Example

Java バージョンの例は参考として samples/doris-demo/spark-demo/ の下に提供されています。こちらをご覧ください。

Configuration

General Configuration

Key	Default Value	Comment
doris.fenodes	--	Doris FE httpアドレス。カンマで区切られた複数のアドレスをサポート
doris.table.identifier	--	Dorisテーブル名（例：db1.tbl1）
doris.user	--	Dorisにアクセスするためのユーザー名
doris.password	Empty string	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	1つのRDD Partitionに対応するDoris Tabletsの数。この値が小さいほど、より多くのPartitionが生成されるため、Sparkの並列性が向上しますが、Dorisにより大きな負荷をかけます。
doris.read.field	--	Dorisテーブルから読み取る列名のリスト。カンマで区切られます
doris.batch.size	4064	BEから一度に読み取る最大行数。この値を増やすと、SparkとDoris間で確立される接続数を減らすことができます。これにより、ネットワーク遅延による追加の時間オーバーヘッドを削減できます。
doris.exec.mem.limit	8589934592	単一クエリのメモリ制限。デフォルトは8GB、バイト単位
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	リトライ回数を設定した後の各リトライ間の間隔（ms単位）。
doris.sink.properties.format	csv	Stream Loadのデータフォーマット。 3つのフォーマットをサポート：csv、json、arrow 詳細パラメータ
doris.sink.properties.*	--	Stream Loadのインポートパラメータ。例：列区切り文字の指定：`'doris.sink.properties.column_separator' = ','` など。詳細パラメータ
doris.sink.task.partition.size	--	Doris書き込みタスクに対応するPartition数。Spark RDDがフィルターやその他の操作を経た後、最終的に書き込まれるPartition数は比較的多くなる可能性がありますが、各Partitionに対応するレコード数は比較的少なく、書き込み頻度の増加と計算リソースの無駄につながります。この値を小さく設定すると、Dorisの書き込み頻度を下げることができ、Dorisのマージ圧力を軽減できます。このパラメータはdoris.sink.task.use.repartitionと組み合わせて使用されます。
doris.sink.task.use.repartition	false	repartition方式を使用してDoris書き込みPartition数を制御するかどうか。デフォルト値はfalseで、coalesce方式を使用して制御します（注意：書き込み前にSpark actionオペレーターがない場合、全体的な計算並列性が低下する可能性があります）。 trueに設定すると、repartition方式が使用されます（注意：最終的なPartition数は設定できますが、追加のshuffleオーバーヘッドが発生します）。
doris.sink.batch.interval.ms	0	各バッチSinkの間隔時間（ms単位）。
doris.sink.enable-2pc	false	二段階コミットを有効にするかどうか。有効にすると、ジョブの最後にトランザクションがコミットされ、一部のタスクが失敗した場合、pre-committed状態のすべてのトランザクションがロールバックされます。
doris.sink.auto-redirect	true	StreamLoadリクエストをリダイレクトするかどうか。有効にすると、StreamLoadは明示的にBE情報を取得せずにFEを通じて書き込みます。
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 and Dataframe Specific Configuration

Key	Default Value	Comment
doris.filter.query.in.max.count	10000	述語プッシュダウンのin式の値リストの最大要素数。この数を超えると、in式条件フィルタリングはSpark側で処理されます。

Structured Streaming Specific Configuration

Key	Default Value	Comment
doris.sink.streaming.passthrough	false	処理せずに最初の列の値を直接書き込みます。

RDD Specific Configuration

Key	Default Value	Comment
doris.request.auth.user	--	Dorisにアクセスするためのユーザー名
doris.request.auth.password	--	Dorisにアクセスするためのパスワード
doris.filter.query	--	読み取りデータをフィルタリングする式。この式はDorisに透過的に送信されます。Dorisはこの式を使用してソースデータフィルタリングを完了します。

Doris to Spark Column Type Mapping

Doris Type	Spark Type
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 to Doris Data Type Mapping

Spark Type	Doris Type
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_RESFUL_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

```sparksql
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_RESFUL_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` は非推奨となり、書き込み時に追加する必要はありません。

:::

2. Overwrite 書き込みの使用方法は？

バージョン 1.3.0 以降、Overwrite モードでの書き込みがサポートされています（テーブル全体レベルのデータ上書きのみサポート）。具体的な使用方法は以下の通りです：

**DataFrame**

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

SQL

```sparksql
INSERT OVERWRITE your_target_table SELECT * FROM your_source_table;
```

3. Bitmap型の読み取り方法

バージョン24.0.0以降、Arrow Flight SQLを通じて変換されたBitmapデータの読み取りがサポートされています（Dorisバージョン >= 2.1.0が必要）。

**BitmapからString**

`DataFrame`メソッドを例にとると、`doris.read.bitmap-to-string`を`true`に設定します。具体的な結果形式については、オプション定義を参照してください。

```scala
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_RESFUL_PORT")
    .option("user", "$YOUR_DORIS_USERNAME")
    .option("password", "$YOUR_DORIS_PASSWORD")
    .option("doris.read.bitmap-to-string", "true")
    .load()
```

Bitmap to Base64

`DataFrame`メソッドを例として、`doris.read.bitmap-to-base64`を`true`に設定します。具体的な結果フォーマットについては、オプション定義を参照してください。

```scala
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_RESFUL_PORT")
    .option("user", "$YOUR_DORIS_USERNAME")
    .option("password", "$YOUR_DORIS_PASSWORD")
    .option("doris.read.bitmap-to-base64", "true")
    .load()
```

4. DataFrame経由での書き込み時のエラー: org.apache.spark.sql.AnalysisException: TableProvider implementation doris cannot be written with ErrorIfExists mode, please use Append or Overwrite modes instead.

保存モードをAppendとして追加する必要があります：

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

バージョン互換性​

使用方法​

Maven​

コンパイル​

使用例​

バッチ読み取り​

RDD​

DataFrame​

Spark SQL​

pySpark​

Arrow Flight SQLによるデータ読み取り​

バッチ書き込み​

DataFrame​

Spark SQL​

ストリーミング書き込み​

DataFrame​

構造化データ書き込み​

Direct Write​

JSON形式での書き込み​

Spark Doris Catalog​

Catalog Config​

DataFrame​

Spark SQL​

Java Example​

Configuration​

General Configuration​

SQL and Dataframe Specific Configuration​

Structured Streaming Specific Configuration​

RDD Specific Configuration​

Doris to Spark Column Type Mapping​

Spark to Doris Data Type Mapping​

FAQ​

バージョン互換性

使用方法

Maven

コンパイル

使用例

バッチ読み取り

RDD

DataFrame

Spark SQL

pySpark

Arrow Flight SQLによるデータ読み取り

バッチ書き込み

DataFrame

Spark SQL

ストリーミング書き込み

DataFrame

構造化データ書き込み

Direct Write

JSON形式での書き込み

Spark Doris Catalog

Catalog Config

DataFrame

Spark SQL

Java Example

Configuration

General Configuration

SQL and Dataframe Specific Configuration

Structured Streaming Specific Configuration

RDD Specific Configuration

Doris to Spark Column Type Mapping

Spark to Doris Data Type Mapping

FAQ