Spark Doris Connector
Spark Doris Connectorは、Sparkを通じてDorisに保存されたデータの読み取りとDorisへのデータの書き込みをサポートできます。
Github: https://github.com/apache/doris-spark-connector
RDD、DataFrame、Spark SQLを通じてDorisからバッチモードでデータを読み取ることをサポートします。DataFrameまたはSpark SQLの使用を推奨します- DataFrame APIとSpark SQLを使用して、バッチモードまたはストリーミングモードで
Dorisにデータを書き込むことをサポートします。 DorisテーブルをDataFrameまたはRDDにマップできます。DataFrameの使用を推奨します。Doris側でのデータフィルタリングの完了をサポートし、データ転送量を削減します。
バージョン互換性
| Connector | Spark | Doris | Java | Scala |
|---|---|---|---|---|
| 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 |
| 24.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.1.0</version>
</dependency>
::: tip
バージョン24.0.0以降、Dorisコネクタパッケージの命名規則が調整されました：
- 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を直接実行できます。詳細については、こちらを参照してください。
コンパイルが成功すると、distディレクトリにターゲットjarパッケージが生成されます。例：spark-doris-connector-spark-3.5-25.1.0.jar。このファイルをSparkのClassPathにコピーしてSpark-Doris-Connectorを使用します。例えば、SparkがLocalモードで実行されている場合、このファイルをjars/フォルダに配置します。SparkがYarnクラスターモードで実行されている場合、このファイルを事前デプロイパッケージに配置します。
また、次のことも可能です
ソースコードディレクトリで実行：
sh build.sh
プロンプトに従って、コンパイルに必要なScalaとSparkバージョンを入力します。
コンパイルが成功すると、distディレクトリにターゲットjarパッケージが生成されます。例：spark-doris-connector-spark-3.5-25.1.0.jar。
このファイルをSparkのClassPathにコピーしてSpark-Doris-Connectorを使用します。
例えば、SparkがLocalモードで実行されている場合、このファイルをjars/フォルダに配置します。SparkがYarnクラスターモードで実行されている場合、このファイルを事前デプロイパッケージに配置します。
例えば、spark-doris-connector-spark-3.5-25.1.0.jarをhdfsにアップロードし、hdfs上のJarパッケージパスをspark.yarn.jarsパラメータに追加します
1. Upload `spark-doris-connector-spark-3.5-25.1.0.jar` to hdfs.
hdfs dfs -mkdir /spark-jars/
hdfs dfs -put /your_local_path/spark-doris-connector-spark-3.5-25.1.0.jar /spark-jars/
2. Add the `spark-doris-connector-spark-3.5-25.1.0.jar` dependency in the cluster.
spark.yarn.jars=hdfs:///spark-jars/spark-doris-connector-spark-3.5-25.1.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 the fields to write
.option("doris.write.fields", "$YOUR_FIELDS_TO_WRITE")
// Support setting Overwrite mode to overwrite data
// .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;
Streaming Write
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()
直接書き込み
データストリーム内のデータの最初の列が、同じ列順序を持つCSV形式のデータや同じフィールド名を持つJSON形式のデータなど、Dorisテーブル構造に準拠したフォーマット済みデータの場合、doris.sink.streaming.passthroughオプションをtrueに設定することで、DataFrameへの変換なしに直接Dorisに書き込むことができます。
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"
);
メッセージの値は {"c0":1,"c1":"a","dt":"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 the value of the message 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_RESFUL_PORT")
.option("user", "$YOUR_DORIS_USERNAME")
.option("password", "$YOUR_DORIS_PASSWORD")
// Set this option to true, and the first column will be written directly without processing.
.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設定
| Key | Required | Comment |
|---|---|---|
| spark.sql.catalog.your_catalog_name | true | catalogプロバイダーのクラス名を設定します。Dorisの場合の唯一の有効な値はorg.apache.doris.spark.catalog.DorisTableCatalogです |
| spark.sql.catalog.your_catalog_name.doris.fenodes | true | fe_ip:fe_http_portの形式でDoris FEノードを設定します |
| spark.sql.catalog.your_catalog_name.doris.query.port | false | Doris FEクエリポートを設定します。spark.sql.catalog.your_catalog_name.doris.fe.auto.fetchがtrueに設定されている場合、このオプションは不要です |
| spark.sql.catalog.your_catalog_name.doris.user | true | Dorisユーザーを設定します |
| spark.sql.catalog.your_catalog_name.doris.password | true | Dorisパスワードを設定します |
| spark.sql.defaultCatalog | false | 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;
設定
一般
| キー | デフォルト値 | コメント |
|---|---|---|
| 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 | 1つのRDD Partitionに対応するDoris Tabletsの数。この値を小さく設定するほど、より多くのパーティションが生成されます。これにより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 | リトライ回数設定後の各リトライ間の間隔、ミリ秒単位 |
| doris.sink.properties.format | -- | stream loadのデータフォーマット。 サポートフォーマット：csv、json、arrow 詳細なマルチパラメータの詳細 |
| doris.sink.properties.* | -- | Stream Loadのインポートパラメータ。 例： 列区切り文字の指定： 'doris.sink.properties.column_separator' = ','。パラメータの詳細 |
| doris.sink.task.partition.size | -- | 書き込みタスクに対応するパーティション数。フィルタリングなどの操作後、Spark RDDで書き込まれるパーティション数は多くなる可能性がありますが、各Partitionに対応するレコード数は比較的少なく、書き込み頻度の増加とコンピューティングリソースの無駄を引き起こします。この値を小さく設定するほど、Doris書き込み頻度が少なくなり、Dorisマージ圧力も少なくなります。一般的にdoris.sink.task.use.repartitionと併用されます。 |
| doris.sink.task.use.repartition | false | Dorisによって書き込まれるパーティション数を制御するためにrepartitionモードを使用するかどうか。デフォルト値はfalseで、coalesceが使用されます（注意：書き込み前にSpark actionがない場合、全体の計算の並列性が低くなります）。trueに設定すると、repartitionが使用されます（注意：shuffleの代償で最終的なパーティション数を設定できます）。 |
| doris.sink.batch.interval.ms | 0 | 各バッチsinkの間隔時間、ミリ秒単位。 |
| doris.sink.enable-2pc | false | 2段階コミットを有効にするかどうか。有効にすると、ジョブ終了時にトランザクションがコミットされ、一部のタスクが失敗した場合はすべてのプリコミットトランザクションがロールバックされます。 |
| doris.sink.auto-redirect | true | StreamLoadリクエストをリダイレクトするかどうか。オンにすると、StreamLoadは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 | 100 | 述語プッシュダウンにおいて、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 |
ヒント
バージョン24.0.0以降、Bitmapタイプの戻り値の型は文字列型で、デフォルトの戻り値は文字列値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
```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"
"doris.ignore-type"="hll",
"doris.write.fields"="col1,hll_col1=hll_hash(col1)"
);
```
ヒント
バージョン24.0.0以降、doris.ignore-typeは非推奨となり、書き込み時にこのパラメータを追加する必要がありません。
-
overwriteを使用して書き込む方法は？
バージョン1.3.0以降、overwriteモードでの書き込みがサポートされています（全テーブルレベルでのデータ上書きのみサポート）。具体的な使用方法は以下の通りです：
DataFrame
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に設定してください。具体的な結果形式については、オプション定義を参照してください。
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に設定してください。具体的な結果形式については、オプション定義を参照してください。
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()
-
DataFrame モードで書き込み時にエラーが発生する:
org.apache.spark.sql.AnalysisException: TableProvider implementation doris cannot be written with ErrorIfExists mode, please use Append or Overwrite modes instead.save mode を append に追加する必要があります。
resultDf.format("doris")
.option("doris.fenodes","$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT")
// your own options
.mode(SaveMode.Append)
.save()