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 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を直接実行できます。詳細については、こちらを参照してください。
コンパイルが成功すると、distディレクトリにターゲットjarパッケージが生成されます。例：spark-doris-connector-spark-3.5-25.1.0.jar。このファイルをSparkのClassPathにコピーしてSpark-Doris-Connectorを使用します。例えば、Localモードで動作するSparkの場合、このファイルをjars/フォルダに配置します。Yarnクラスターモードで動作するSparkの場合、このファイルを事前デプロイメントパッケージに配置します。
また、以下も可能です
ソースコードディレクトリで実行：
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()
直接書き込み
データストリーム内のデータの最初の列がDorisテーブル構造に準拠したフォーマット済みデータ（同じ列順序のCSVフォーマットデータや同じフィールド名のJSONフォーマットデータなど）の場合、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"
);
messageの値は{"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 Config
| 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;
設定
全般
| Key | デフォルト値 | コメント |
|---|---|---|
| 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の数。この値を小さく設定するほど、より多くのパーティションが生成される。これにより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 | -- | 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アクションがない場合、全体的な計算の並列性が低くなる）。trueに設定すると、repartitionが使用される（注意：shuffleのコストで最終パーティション数を設定できる） |
| doris.sink.batch.interval.ms | 0 | 各バッチsinkの間隔時間、単位はms |
| 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設定
| Key | デフォルト値 | コメント |
|---|---|---|
| doris.filter.query.in.max.count | 100 | 述語プッシュダウンにおいて、in式の値リストの最大要素数。この数を超えた場合、in式の条件フィルタリングはSpark側で処理される |
Structured Streaming設定
| Key | デフォルト値 | コメント |
|---|---|---|
| doris.sink.streaming.passthrough | false | 処理を行わずに最初の列の値を直接書き込む |
RDD設定
| Key | デフォルト値 | コメント |
|---|---|---|
| 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()