メインコンテンツまでスキップ

Arrow Flight SQLプロトコルによる接続

Doris 2.1以降、Arrow Flight SQLプロトコルに基づく高速データリンクが実装され、SQLクエリを使用してDorisから大容量データを複数の言語で高速に取得できるようになりました。Arrow Flight SQLはユニバーサルJDBCドライバも提供し、Arrow Flight SQLプロトコルに準拠する他のデータベースとのシームレスな連携をサポートします。一部のシナリオでは、MySQL ClientやJDBC/ODBCドライバを使用するデータ転送ソリューションと比較して、パフォーマンスが最大100倍向上することがあります。

実装原理​

Dorisでは、クエリ結果はBlocksとして列形式で構成されます。2.1以前のバージョンでは、データはMySQL ClientやJDBC/ODBCドライバを介してターゲットクライアントに転送できましたが、これには行ベースのBytesを列形式に逆シリアル化する必要がありました。Arrow Flight SQLベースの高速データ転送リンクを構築することで、ターゲットクライアントもArrow列形式をサポートしている場合、転送プロセス全体でシリアル化および逆シリアル化操作が回避され、それらに関連する時間とパフォーマンスのオーバーヘッドが完全に排除されます。

Arrow Flight SQL

Apache Arrowをインストールするには、公式ドキュメントApache Arrowで詳細なインストール手順を確認できます。DorisがArrow Flightプロトコルを実装する方法の詳細については、Doris support Arrow Flight SQL protocolを参照してください。

Python使用方法​

PythonのADBC​​Driverを使用してDorisに接続し、非常に高速なデータ読み取りを実現します。以下の手順では、Python（バージョン >= 3.9）ADBC​​Driverを使用して、DDL、DML、Session変数の設定、Show文を含む一連の一般的なデータベース構文操作を実行します。

ライブラリのインストール​

ライブラリはPyPIで公開されており、以下の方法で簡単にインストールできます：

pip install adbc_driver_manager
pip install adbc_driver_flightsql

インストールされたライブラリを使用するために、コード内で以下のモジュール/ライブラリをインポートしてください：

import adbc_driver_manager
import adbc_driver_flightsql.dbapi as flight_sql

>>> print(adbc_driver_manager.__version__)
1.1.0
>>> print(adbc_driver_flightsql.__version__)
1.1.0

Dorisへの接続​

Doris Arrow Flight SQLサービスと対話するためのクライアントを作成します。Doris FEのHost、Arrow Flight Port、ログインユーザー名とパスワードを提供し、以下の設定を実行する必要があります。 Doris FEとBEの設定パラメータを変更します：

  • fe/conf/fe.confのarrow_flight_sql_portを8070など利用可能なポートに変更します。
  • be/conf/be.confのarrow_flight_sql_portを8050など利用可能なポートに変更します。

注意: fe.confとbe.confで設定されるarrow_flight_sql_portのポート番号は異なります

設定を変更してクラスターを再起動した後、fe/log/fe.logファイルでArrow Flight SQL service is startedを検索すると、FEのArrow Flight Serverが正常に開始されたことを示します；be/log/be.INFOファイルでArrow Flight Service bind to hostを検索すると、BEのArrow Flight Serverが正常に開始されたことを示します。

DorisインスタンスのFEとBEのArrow Flight SQLサービスがそれぞれポート8070と8050で実行され、Dorisのユーザー名/パスワードが"user"/"pass"であると仮定すると、接続プロセスは以下の通りです：

conn = flight_sql.connect(uri="grpc://{FE_HOST}:{fe.conf:arrow_flight_sql_port}", db_kwargs={
            adbc_driver_manager.DatabaseOptions.USERNAME.value: "user",
            adbc_driver_manager.DatabaseOptions.PASSWORD.value: "pass",
        })
cursor = conn.cursor()

接続が完了すると、返されたCursorを使用してSQLを通じてDorisと対話し、テーブルの作成、メタデータの取得、データのインポート、クエリなどの操作を実行できます。

テーブルの作成とメタデータの取得​

cursor.execute()関数にQueryを渡して、テーブル作成とメタデータ取得操作を実行します：

cursor.execute("DROP DATABASE IF EXISTS arrow_flight_sql FORCE;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("create database arrow_flight_sql;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("show databases;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("use arrow_flight_sql;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("""CREATE TABLE arrow_flight_sql_test
    (
         k0 INT,
         k1 DOUBLE,
         K2 varchar(32) NULL DEFAULT "" COMMENT "",
         k3 DECIMAL(27,9) DEFAULT "0",
         k4 BIGINT NULL DEFAULT '10',
         k5 DATE,
    )
    DISTRIBUTED BY HASH(k5) BUCKETS 5
    PROPERTIES("replication_num" = "1");""")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("show create table arrow_flight_sql_test;")
print(cursor.fetchallarrow().to_pandas())

StatusResultが0を返す場合、Queryが正常に実行されたことを意味します（この設計の理由はJDBCとの互換性を保つためです）。

  StatusResult
0            0

  StatusResult
0            0

                   Database
0         __internal_schema
1          arrow_flight_sql
..                      ...
507             udf_auth_db

[508 rows x 1 columns]

  StatusResult
0            0

  StatusResult
0            0
                   Table                                       Create Table
0  arrow_flight_sql_test  CREATE TABLE `arrow_flight_sql_test` (\n  `k0`...

データのインポート​

作成されたテーブルに少量のテストデータをインポートするため、INSERT INTOを実行します：

cursor.execute("""INSERT INTO arrow_flight_sql_test VALUES
        ('0', 0.1, "ID", 0.0001, 9999999999, '2023-10-21'),
        ('1', 0.20, "ID_1", 1.00000001, 0, '2023-10-21'),
        ('2', 3.4, "ID_1", 3.1, 123456, '2023-10-22'),
        ('3', 4, "ID", 4, 4, '2023-10-22'),
        ('4', 122345.54321, "ID", 122345.54321, 5, '2023-10-22');""")
print(cursor.fetchallarrow().to_pandas())

以下により、インポートが成功したことが証明されます：

  StatusResult
0            0

大量のデータをDorisにインポートする必要がある場合は、pydorisを使用してStream Loadを実行できます。

クエリの実行​

次に、上記でインポートしたテーブルをクエリします。これには集約、ソート、Set Session Variableなどの操作が含まれます。

cursor.execute("select * from arrow_flight_sql_test order by k0;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("set exec_mem_limit=2000;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("show variables like \"%exec_mem_limit%\";")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("select k5, sum(k1), count(1), avg(k3) from arrow_flight_sql_test group by k5;")
print(cursor.fetch_df())

結果は以下の通りです：

   k0            k1    K2                k3          k4          k5
0   0       0.10000    ID       0.000100000  9999999999  2023-10-21
1   1       0.20000  ID_1       1.000000010           0  2023-10-21
2   2       3.40000  ID_1       3.100000000      123456  2023-10-22
3   3       4.00000    ID       4.000000000           4  2023-10-22
4   4  122345.54321    ID  122345.543210000           5  2023-10-22

[5 rows x 6 columns]

  StatusResult
0            0

    Variable_name Value Default_Value Changed
0  exec_mem_limit  2000    2147483648       1

           k5  Nullable(Float64)_1  Int64_2 Nullable(Decimal(38, 9))_3
0  2023-10-22         122352.94321        3            40784.214403333
1  2023-10-21              0.30000        2                0.500050005

[2 rows x 5 columns]

注意: クエリ結果を取得するには、cursor.fetchallarrow()を使用してarrow形式を返すか、cursor.fetch_df()を使用してpandas dataframeを直接返す必要があります。これによりデータが列形式で保持されます。cursor.fetchall()は使用しないでください。そうしないと列形式のデータが行形式に変換され、本質的にはmysql-clientを使用するのと同じになってしまいます。実際、クライアント側での余分な列から行への変換操作により、mysql-clientよりも遅くなる可能性があります。

完全なコード​

# Doris Arrow Flight SQL Test

# step 1, library is released on PyPI and can be easily installed.
# pip install adbc_driver_manager
# pip install adbc_driver_flightsql
import adbc_driver_manager
import adbc_driver_flightsql.dbapi as flight_sql

# step 2, create a client that interacts with the Doris Arrow Flight SQL service.
# Modify arrow_flight_sql_port in fe/conf/fe.conf to an available port, such as 8070.
# Modify arrow_flight_sql_port in be/conf/be.conf to an available port, such as 8050.
conn = flight_sql.connect(uri="grpc://{FE_HOST}:{fe.conf:arrow_flight_sql_port}", db_kwargs={
            adbc_driver_manager.DatabaseOptions.USERNAME.value: "root",
            adbc_driver_manager.DatabaseOptions.PASSWORD.value: "",
        })
cursor = conn.cursor()

# interacting with Doris via SQL using Cursor
def execute(sql):
    print("\n### execute query: ###\n " + sql)
    cursor.execute(sql)
    print("### result: ###")
    print(cursor.fetchallarrow().to_pandas())

# step3, execute DDL statements, create database/table, show stmt.
execute("DROP DATABASE IF EXISTS arrow_flight_sql FORCE;")
execute("show databases;")
execute("create database arrow_flight_sql;")
execute("show databases;")
execute("use arrow_flight_sql;")
execute("""CREATE TABLE arrow_flight_sql_test
    (
         k0 INT,
         k1 DOUBLE,
         K2 varchar(32) NULL DEFAULT "" COMMENT "",
         k3 DECIMAL(27,9) DEFAULT "0",
         k4 BIGINT NULL DEFAULT '10',
         k5 DATE,
    )
    DISTRIBUTED BY HASH(k5) BUCKETS 5
    PROPERTIES("replication_num" = "1");""")
execute("show create table arrow_flight_sql_test;")


# step4, insert into
execute("""INSERT INTO arrow_flight_sql_test VALUES
        ('0', 0.1, "ID", 0.0001, 9999999999, '2023-10-21'),
        ('1', 0.20, "ID_1", 1.00000001, 0, '2023-10-21'),
        ('2', 3.4, "ID_1", 3.1, 123456, '2023-10-22'),
        ('3', 4, "ID", 4, 4, '2023-10-22'),
        ('4', 122345.54321, "ID", 122345.54321, 5, '2023-10-22');""")


# step5, execute queries, aggregation, sort, set session variable
execute("select * from arrow_flight_sql_test order by k0;")
execute("set exec_mem_limit=2000;")
execute("show variables like \"%exec_mem_limit%\";")
execute("select k5, sum(k1), count(1), avg(k3) from arrow_flight_sql_test group by k5;")

# step6, close cursor 
cursor.close()

Jdbc Connector with Arrow Flight SQL​

Arrow Flight SQL プロトコルのオープンソース JDBC ドライバーは、標準 JDBC API と互換性があり、ほとんどの BI ツールが JDBC を通じて Doris にアクセスするために使用でき、Apache Arrow データの高速転送をサポートします。使用方法は MySQL プロトコルの JDBC ドライバーを通じて Doris に接続する場合と似ています。リンク URL の jdbc:mysql プロトコルを jdbc:arrow-flight-sql プロトコルに置き換えるだけです。クエリ結果は引き続き JDBC ResultSet データ構造で返されます。

POM dependency:

<properties>
    <arrow.version>17.0.0</arrow.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.apache.arrow</groupId>
        <artifactId>flight-sql-jdbc-core</artifactId>
        <version>${arrow.version}</version>
    </dependency>
</dependencies>

注意: Java 9以降を使用する場合、Javaコマンドに--add-opens=java.base/java.nio=ALL-UNNAMEDを追加してJDKの内部構造を公開する必要があります。そうしないと、module java.base does not "opens java.nio" to unnamed moduleやmodule java.base does not "opens java.nio" to org.apache.arrow.memory.core、java.lang.NoClassDefFoundError: Could not initialize class org.apache.arrow.memory.util.MemoryUtil (Internal; Prepare)などのエラーが発生する可能性があります。

# Directly on the command line
$ java --add-opens=java.base/java.nio=ALL-UNNAMED -jar ...
# Indirectly via environment variables
$ env _JAVA_OPTIONS="--add-opens=java.base/java.nio=ALL-UNNAMED" java -jar ...

IntelliJ IDEAでデバッグする場合は、Run/Debug ConfigurationsのBuild and runに--add-opens=java.base/java.nio=ALL-UNNAMEDを追加する必要があります。下図を参照してください：

arrow-flight-sql-IntelliJ

接続コードの例は以下の通りです：

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

Class.forName("org.apache.arrow.driver.jdbc.ArrowFlightJdbcDriver");
String DB_URL = "jdbc:arrow-flight-sql://{FE_HOST}:{fe.conf:arrow_flight_sql_port}?useServerPrepStmts=false"
        + "&cachePrepStmts=true&useSSL=false&useEncryption=false";
String USER = "root";
String PASS = "";

Connection conn = DriverManager.getConnection(DB_URL, USER, PASS);
Statement stmt = conn.createStatement();
ResultSet resultSet = stmt.executeQuery("select * from information_schema.tables;");
while (resultSet.next()) {
    System.out.println(resultSet.toString());
}

resultSet.close();
stmt.close();
conn.close();

Java使用方法​

JDBCの使用に加えて、Pythonと同様に、JAVAでもDriverを作成してDorisを読み取り、Arrow形式でデータを返すことができます。以下は、AdbcDriverとJdbcDriverを使用してDoris Arrow Flight Serverに接続する方法です。

POM依存関係：

<properties>
    <adbc.version>0.15.0</adbc.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-driver-jdbc</artifactId>
        <version>${adbc.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-core</artifactId>
        <version>${adbc.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-driver-manager</artifactId>
        <version>${adbc.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-sql</artifactId>
        <version>${adbc.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-driver-flight-sql</artifactId>
        <version>${adbc.version}</version>
    </dependency>
</dependencies>

ADBC Driver​

接続コードの例は以下の通りです：

// 1. new driver
final BufferAllocator allocator = new RootAllocator();
FlightSqlDriver driver = new FlightSqlDriver(allocator);
Map<String, Object> parameters = new HashMap<>();
AdbcDriver.PARAM_URI.set(parameters, Location.forGrpcInsecure("{FE_HOST}", {fe.conf:arrow_flight_sql_port}).getUri().toString());
AdbcDriver.PARAM_USERNAME.set(parameters, "root");
AdbcDriver.PARAM_PASSWORD.set(parameters, "");
AdbcDatabase adbcDatabase = driver.open(parameters);

// 2. new connection
AdbcConnection connection = adbcDatabase.connect();
AdbcStatement stmt = connection.createStatement();

// 3. execute query
stmt.setSqlQuery("select * from information_schema.tables;");
QueryResult queryResult = stmt.executeQuery();
ArrowReader reader = queryResult.getReader();

// 4. load result
List<String> result = new ArrayList<>();
while (reader.loadNextBatch()) {
    VectorSchemaRoot root = reader.getVectorSchemaRoot();
    String tsvString = root.contentToTSVString();
    result.add(tsvString);
}
System.out.printf("batchs %d\n", result.size());

// 5. close
reader.close();
queryResult.close();
stmt.close();
connection.close();

JDBC Driver​

Java 9以降を使用する場合、java コマンドに --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED を追加して、一部のJDK内部を公開する必要があります：

# Directly on the command line
$ java --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED -jar ...
# Indirectly via environment variables
$ env _JAVA_OPTIONS="--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED" java -jar ...

そうでなければ、module java.base does not "opens java.nio" to unnamed module や module java.base does not "opens java.nio" to org.apache.arrow.memory.core または ava.lang.NoClassDefFoundError: Could not initialize class org.apache.arrow.memory.util.MemoryUtil (Internal; Prepare) などのエラーが表示される場合があります。

IntelliJ IDEAでデバッグする場合は、Run/Debug ConfigurationsのBuild and runで--add-opens=java.base/java.nio=ALL-UNNAMEDを追加する必要があります。以下の画像を参照してください：

IntelliJ IDEA

接続コードの例は以下の通りです：

final Map<String, Object> parameters = new HashMap<>();
AdbcDriver.PARAM_URI.set(
        parameters,"jdbc:arrow-flight-sql://{FE_HOST}:{fe.conf:arrow_flight_sql_port}?useServerPrepStmts=false&cachePrepStmts=true&useSSL=false&useEncryption=false");
AdbcDriver.PARAM_USERNAME.set(parameters, "root");
AdbcDriver.PARAM_PASSWORD.set(parameters, "");
try (
        BufferAllocator allocator = new RootAllocator();
        AdbcDatabase db = new JdbcDriver(allocator).open(parameters);
        AdbcConnection connection = db.connect();
        AdbcStatement stmt = connection.createStatement()
) {
    stmt.setSqlQuery("select * from information_schema.tables;");
    AdbcStatement.QueryResult queryResult = stmt.executeQuery();
    ArrowReader reader = queryResult.getReader();
    List<String> result = new ArrayList<>();
    while (reader.loadNextBatch()) {
        VectorSchemaRoot root = reader.getVectorSchemaRoot();
        String tsvString = root.contentToTSVString();
        result.add(tsvString);
    }
    long etime = System.currentTimeMillis();
    System.out.printf("batchs %d\n", result.size());

    reader.close();
    queryResult.close();
    stmt.close();
}  catch (Exception e) {
    e.printStackTrace();
}

JdbcとJava接続方式の選択​

JDBC/Java Arrow Flight SQL Sampleは、Arrow FLight SQLを使用したJDBC/Javaのデモです。これを使用してArrow Flight Serverへクエリを送信するさまざまな接続方式をテストでき、Arrow FLight SQLの使用方法の理解とパフォーマンステストに役立ちます。期待される実行結果については、Add Arrow Flight Sql demo for Javaを参照してください。

従来のjdbc:mysql接続方式と比較したJdbcとJavaのArrow Flight SQL接続方式のパフォーマンステストは、GitHub Issueのセクション6.2で確認できます。テスト結論に基づいた使用提案をいくつか示します。

  1. 上記3つのJava Arrow Flight SQL接続方式について、後続のデータ分析が行ベースのデータ形式に基づく場合は、JDBC ResultSet形式でデータを返すjdbc:arrow-flight-sqlを使用してください。後続のデータ分析がArrow形式またはその他の列ベースのデータ形式に基づくことができる場合は、Flight AdbcDriverまたはFlight JdbcDriverを使用してArrow形式で直接データを返すことで、行列変換を回避し、Arrowの特性を使用してデータ解析を高速化できます。

  2. JDBC ResultSetまたはArrow形式でのデータ解析のどちらでも、費やされる時間はデータ読み取りに費やされる時間より長くなります。Arrow Flight SQLのパフォーマンスが期待通りでなく、jdbc:mysql://と比較して改善が限定的な場合は、データ解析に時間がかかりすぎているかを分析することをお勧めします。

  3. すべての接続方式において、JDK 17はJDK 1.8よりもデータ読み取りが高速です。

  4. 大量のデータを読み取る際、Arrow Flight SQLはjdbc:mysql://よりも少ないメモリを使用するため、メモリ不足に悩まされている場合は、Arrow Flight SQLを試すこともできます。

  5. 上記3つの接続方式に加えて、ネイティブのFlightClientを使用してArrow Flight Serverに接続することもでき、複数のエンドポイントをより柔軟に並列で読み取ることができます。Flight AdbcDriverもFlightClientに基づいて作成されたリンクで、FlightClientを直接使用するよりも簡単です。

他のビッグデータコンポーネントとの連携​

Spark & Flink​

Arrow Flightには現在、SparkとFlinkをサポートする公式計画はありません（GitHub Issue）。バージョン24.0.0以降、Doris独自のSpark ConnectorとFlink ConnectorはArrow Flight SQL経由でのDorisアクセスをサポートしており、これによりリードパフォーマンスが数倍向上することが期待されます。

コミュニティでは以前、オープンソースのSpark-Flight-Connectorを参照し、SparkでFlightClientを使用してDorisに接続してテストを行いました。その結果、ArrowとDoris Block間のデータ形式変換が高速であることが判明し、CSV形式とDoris Block間の変換速度の10倍で、MapやArrayなどの複合型により良いサポートを提供します。これは、Arrow データ形式が高い圧縮率を持ち、送信時のネットワークオーバーヘッドが低いためです。ただし、Doris Arrow Flightはまだマルチノード並列読み取りを実装しておらず、依然としてクエリ結果をBEノードに集約して返します。単純なデータのバッチエクスポートでは、Tablet レベルの並列読み取りをサポートするDoris Spark Connectorほど高速ではない可能性があります。SparkでArrow Flight SQLを使用してDorisに接続したい場合は、オープンソースのSpark-Flight-ConnectorとDremio-Flight-Connectorを参照して自身で実装できます。

BIツールのサポート​

Doris v2.1.8以降、DBeaverなどのBIツールがarrow-flight-sqlプロトコルを使用してDorisに接続することをサポートしています。DBeaverのarrow-flight-sql DriverでDorisに接続する方法については、以下を参照してください：how-to-use-jdbc-driver-with-dbeaver-client、client-applications/clients/dbeaver/。

拡張アプリケーション​

複数BEの並列結果返却​

Dorisはデフォルトで、すべてのBEノードでのクエリ結果を1つのBEノードに集約します。Mysql/JDBCクエリでは、FEがこの集約されたデータノードからクエリ結果を要求します。Arrow Flight SQLクエリでは、FEがこのノードのIP/PortをEndpointでラップしてADBC Clientに返します。ADBC ClientはこのEndpointに対応するBEノードにリクエストを送信してデータをプルします。

クエリがDorisからデータをプルする単純なSelectで、Join、Sort、Window Functionなどのデータシャッフル動作を持つオペレーターがない場合、クエリはTablet粒度で分割できます。現在、Doris Spark/Flink Connectorはこの方式を使用して並列データ読み取りを実装しており、2つのステップに分かれています：

  1. explain sqlを実行し、FEが返すクエリプランのScanOperatorがScanのすべてのTablet IDリストを含む。
  2. 上記のTablet IDリストに基づいて元のSQLを複数のSQLに分割する。各SQLは一部のTabletのみを読み取る。使用法はSELECT * FROM t1 TABLET(10001,10002) limit 1000;に似ている。分割後の複数のSQLは並列実行できる。Support select table sampleを参照。

クエリの最外層が集約の場合、SQLはselect k1, sum(k2) from xxx group by k1に似ています。Doris v3.0.4以降、set enable_parallel_result_sink=true;を実行することで、クエリの各BEノードが独立してクエリ結果を返すことを許可します。FEが返すEndpointリストを受信後、ADBC Clientは複数のBEノードから並列で結果をプルします。ただし、集約結果が非常に小さい場合、複数のBEを返すとRPCの圧力が増加することに注意してください。具体的な実装については、support parallel result sinkを参照してください。理論的には、最外層のクエリがソートされている場合を除き、他のクエリは各BEノードが並列で結果を返すことをサポートできますが、現在この利便性の必要性はなく、さらなる実装は行われていません。

複数BEがクラスター外からアクセス可能な同一IPを共有​

DorisクラスターがありFEノードはクラスター外からアクセス可能だが、すべてのBEノードはクラスター内からのみアクセス可能な場合があります。Mysql ClientとJDBCを使用してDorisに接続してクエリを実行する場合は問題ありませんが、クエリ結果はDoris FEノードによって返されます。しかし、Arrow Flight SQLを使用してDorisに接続する場合は機能しません。ADBC ClientがDoris BEノードに接続してクエリ結果をプルする必要がありますが、Doris BEノードはクラスター外からのアクセスが許可されていないためです。

本番環境では、Doris BEノードをクラスター外に公開することは不便な場合が多くあります。しかし、すべてのDoris BEノードにリバースプロキシ（Nginxなど）を追加できます。クラスター外のクライアントがNginxに接続すると、Doris BEノードにランダムにルーティングされます。デフォルトでは、Arrow Flight SQLクエリ結果はDoris BEノードにランダムに保存されます。NginxによってランダムにルーティングされたDoris BEノードと異なる場合、Doris BEノード内でのデータ転送が必要になります。

Doris v2.1.8以降、すべてのDoris BEノードのbe.confでpublic_hostとarrow_flight_sql_proxy_portを、複数のDoris BEノードが共有しクラスター外からアクセス可能なIPとポートに設定することで、クエリ結果を正しく転送してADBC Clientに返すことができます。

public_host={nginx ip}
arrow_flight_sql_proxy_port={nginx port}

FAQ​

  1. Q: エラー connection error: desc = "transport: Error while dialing: dial tcp <ip:arrow_flight_port>: i/o timeout"。

A: エラーメッセージの<ip:arrow_flight_port>がDoris FEノードのIPとarrow-flight-portの場合、

まずDoris FEノードのarrow-flight-serverが正常に開始されているかを確認してください。fe/log/fe.logファイル内でArrow Flight SQL service is startedを検索すると、FEのArrow Flight Serverが正常に開始されていることを示します。

Doris FEノードのarrow-flight-serverが正常に開始されている場合、Clientが存在するマシンがエラーメッセージ内のIP <ip:arrow_flight_port>にpingできるかを確認してください。pingできない場合は、Doris FEノードに外部からアクセス可能なIPを開放し、クラスターを再デプロイする必要があります。

A: エラーメッセージの<ip:arrow_flight_port>がDoris BEノードのIPとarrow-flight-portの場合。

まずDoris BEノードのarrow-flight-serverが正常に開始されているかを確認してください。be/log/be.INFOファイル内でArrow Flight Service bind to hostを検索すると、BEのArrow Flight Serverが正常に開始されていることを示します。

Doris BEノードのarrow-flight-serverが正常に開始されている場合、clientマシンがエラーメッセージで報告された<ip:arrow_flight_port>内のIPにpingできるかを確認してください。pingできない場合、Doris BEノードが外部からアクセスできないイントラネット内にあることが判明している場合は、以下の2つの方法を使用します：

  • 各Doris BEノードに外部からアクセス可能なIPを開放することを検討してください。Doris v2.1.8以降、このDoris BEノードのbe.confでこのIPにpublic_hostを設定できます。同様に、すべてのDoris BEノードのpublic_hostを、clientがアクセスできる対応するBEノードのIPに設定してください。

  • 上記のセクション[複数のBEがクラスターで外部からアクセス可能な同一IPを共有]を参照して、すべてのDoris BEノードにリバースプロキシのレイヤーを追加してください。

Doris BEが完全にイントラネット内にあるかどうかが不明な場合は、clientマシンとDoris BEノードが存在するマシンの他のIPとの接続性を確認してください。Doris BEノードが存在するマシンでifconfigを実行して、現在のマシンのすべてのIPを返します。IPの1つは<ip:arrow_flight_port>内のIPと同じであり、show backendsで印刷されるDoris BEノードのIPと同じである必要があります。ifconfigによって返された他のIPを順番にpingしてください。Doris BEノードにClientがアクセスできるIPがある場合は、上記を参照してこのIPをpublic_hostとして設定してください。Doris BEノードのすべてのIPがClientからアクセスできない場合、そのDoris BEノードは完全にイントラネット内にあります。

  1. Q: JDBCまたはJAVAを使用してArrow Flight SQLに接続する際に、エラーメッセージが表示される：module java.base does not "opens java.nio" to unnamed moduleまたはmodule java.base does not "opens java.nio" to org.apache.arrow.memory.coreまたはjava.lang.NoClassDefFoundError: Could not initialize class org.apache.arrow.memory.util.MemoryUtil (Internal; Prepare)

A: まずfe/conf/fe.conf内のJAVA_OPTS_FOR_JDK_17に--add-opens=java.base/java.nio=ALL-UNNAMEDが含まれているかを確認してください。含まれていない場合は追加してください。次に、上記の[JDBC Connector with Arrow Flight SQL]の注記を参照し、Javaコマンドに--add-opens=java.base/java.nio=ALL-UNNAMEDを追加してください。IntelliJ IDEAでデバッグする場合は、Run/Debug ConfigurationsのBuild and runに--add-opens=java.base/java.nio=ALL-UNNAMEDを追加する必要があります。

  1. Q: ARM環境でエラーget flight info statement failed, arrow flight schema timeout, TimeoutException: Waited 5000 milliseconds for io.grpc.stub.Clientが報告される。

A: Linuxカーネルバージョンが<= 4.19.90の場合、4.19.279以上にアップグレードするか、より低いバージョンのLinuxカーネル環境でDoris BEを再コンパイルする必要があります。具体的なコンパイル方法については、ドキュメント<docs/dev/install/source-install/compilation-arm>を参照してください。

原因：これは古いバージョンのLinuxカーネルとArrowの間に互換性の問題があるためです。cpp: arrow::RecordBatch::MakeEmpty()がArrow Record Batchを構築する際にスタックし、Doris BEのArrow Flight ServerがDoris FEのArrow Flight ServerのRPCリクエストに5000ms以内に応答できなくなり、FEがClientにrpc timeout failedを返します。SparkとFlinkがDorisを読み取る際にも、クエリ結果をArrow Record Batchに変換して返すため、同じ問題が存在します。

kylinv10 SP2およびSP3のLinuxカーネルバージョンは最大でも4.19.90-24.4.v2101.ky10.aarch64のみです。カーネルバージョンをさらにアップグレードすることはできません。Doris BEをkylinv10で再コンパイルすることしかできません。新しいバージョンのldb_toolchainでDoris BEをコンパイルしても問題が続く場合は、より低いバージョンのldb_toolchain v0.17でのコンパイルを試すことができます。ARM環境が外部ネットワークに接続できない場合、Huawei CloudはARM + kylinv10を提供し、Alibaba CloudはX86 + kylinv10を提供しています。

  1. Q: Prepared statementでパラメータを渡すとエラーが報告される。

A: 現在、jdbc:arrow-flight-sqlとJava ADBC/JDBCDriverはprepared statementのパラメータ渡しをサポートしていません。例えば、select * from xxx where id=?はparameter ordinal 1 out of rangeエラーを報告します。これはArrow Flight SQLのバグです（GitHub Issue）。

  1. Q: 一部のシナリオでパフォーマンスを向上させるために、jdbc:arrow-flight-sqlが毎回読み取るバッチサイズを変更する方法。

A: org.apache.arrow.adbc.driver.jdbc.JdbcArrowReaderファイルのmakeJdbcConfigメソッド内のsetTargetBatchSizeを変更することで、デフォルトは1024です。変更後のファイルを同じパス名でローカルディレクトリに保存し、元のファイルを上書きして有効にします。

  1. Q: ADBC v0.10、JDBCおよびJava ADBC/JDBCDriverは並列読み取りをサポートしていません。

A: stmt.executePartitioned()メソッドが実装されていません。ネイティブFlightClientを使用して複数のエンドポイントの並列読み取りを実装することしかできません。方法はsqlClient=new FlightSqlClient, execute=sqlClient.execute(sql), endpoints=execute.getEndpoints(), for(FlightEndpoint endpoint: endpoints)です。また、ADBC V0.10のデフォルトAdbcStatementは実際にはJdbcStatementです。executeQuery後、行形式のJDBC ResultSetがArrowカラム形式に変換されます。Java ADBCがADBC 1.0.0で完全に機能することが期待されていますGitHub Issue。

  1. Q: URLでデータベース名を指定する。

A: Arrow v15.0現在、Arrow JDBC ConnectorはURLでのデータベース名の指定をサポートしていません。例えば、jdbc:arrow-flight-sql://{FE_HOST}:{fe.conf:arrow_flight_sql_port}/test?useServerPrepStmts=falseでtestデータベースへの接続を指定しても無効で、手動でSQL use databaseを実行する必要があります。Arrow v18.0はURLでのデータベース名指定をサポートしていますが、実際のテストではまだバグがあります。

  1. Q: Python ADBCがWarning: Cannot disable autocommit; conn will not be DB-API 2.0 compliantを印刷する。

A: Pythonを使用する際はこのWarningを無視してください。これはPython ADBC Clientの問題で、クエリには影響しません。

  1. Q: Pythonでエラーgrpc: received message larger than max (20748753 vs. 16777216)が報告される。

A: Python: grpc: received message larger than max (20748753 vs. 16777216) #2078を参照し、Database Optionにadbc_driver_flightsql.DatabaseOptions.WITH_MAX_MSG_SIZE.valueを追加してください。

  1. Q: エラーinvalid bearer tokenが報告される。

A: SET PROPERTY FOR 'root' 'max_user_connections' = '10000';を実行して、現在のユーザーの最大接続数を10000に変更し、fe.confにqe_max_connection=30000とarrow_flight_token_cache_size=8000を追加してFEを再起動してください。

ADBC ClientとArrow Flight Server間の接続は本質的に長期リンクで、Auth Token、Connection、SessionをServerでキャッシュする必要があります。接続作成後、単一クエリの終了時に即座に切断されることはありません。Clientがclose()リクエストを送信してクリーンアップする必要がありますが、実際にはClientがcloseリクエストを送信しないことが多く、Auth Token、Connection、SessionがArrow Flight Serverに長期間保存されます。デフォルトでは3日後にタイムアウトして切断されるか、接続数がarrow_flight_token_cache_sizeの制限を超えた後にLRUに従って削除されます。

Doris v2.1.8現在、Arrow Flight接続とMysql/JDBC接続は同じ接続制限を使用し、すべてのFEユーザーの総接続数qe_max_connectionとUserProperty内の単一ユーザーの接続数max_user_connectionsが含まれます。しかし、デフォルトのqe_max_connectionとmax_user_connectionsはそれぞれ1024と100です。Arrow Flight SQLはしばしばJDBCシナリオを置き換えるために使用されますが、JDBC接続はクエリ終了後に即座に解放されます。したがって、Arrow Flight SQLを使用する際は、Dorisのデフォルト接続制限が小さすぎるため、接続数がarrow_flight_token_cache_sizeの制限を超え、使用中の接続が削除されることがよくあります。

  1. Q: JDBCまたはJAVAを使用してArrow Flight SQLに接続してDatatime型を読み取ると、フォーマットされた時間ではなくタイムスタンプが返される。

A: JDBCまたはJAVAを使用してArrow Flight SQLに接続してDatatime型を読み取る場合は、自分でタイムスタンプを変換する必要があります。Add java parsing datetime type in arrow flight sql sample #48578を参照してください。Python Arrow Flight SQLを使用してDatatime型を読み取ると2025-03-03 17:23:28Zの結果が返されますが、JDBCまたはJAVAは1740993808を返します。

  1. Q: JDBCまたはJava JDBC Clientを使用してArrow Flight SQLに接続してArray入れ子型を読み取るとエラーConfiguration does not provide a mapping for array column 2が返される。

A: sample/arrow-flight-sqlを参照してJAVA ADBC Clientを使用してください。

Python ADBC Client、JAVA ADBC Client、Java JDBC DriverManagerはすべてArray入れ子型の読み取りに問題ありません。JDBCまたはJava JDBC ClientでArrow Flight SQLに接続する場合のみ問題があります。実際、Arrow Flight JDBCの互換性は保証されていません。これはArrowによって公式に開発されたものではなく、サードパーティデータベース会社Dremioによるものです。以前にも他の互換性問題が見つかっているため、まずJAVA ADBC Clientの使用を推奨します。

2.1 リリースノート​

Doris Arrow Flightはv2.1.4以前のバージョンでは完全ではないため、使用前にアップグレードすることを推奨します。

v2.1.9​

  1. DorisデータのArrowへのシリアライゼーション問題を修正。 Fix UT DataTypeSerDeArrowTest of Array/Map/Struct/Bitmap/HLL/Decimal256 types
  • Decimal256型の読み取りに失敗；
  • DatetimeV2型の読み取りで微細なエラー；
  • DateV2型の読み取りで結果が正しくない；
  • IPV4/IPV6型の読み取りで結果がNULLの場合のエラー；
  1. Doris Arrow Flight SQLクエリが失敗して空の結果を返し、実際のエラー情報を返さない問題を修正。 Fix query result is empty and not return query error message

v2.1.8​

  1. DBeaverなどのBIツールがarrow-flight-sqlプロトコルを使用してDorisに接続することをサポートし、メタデータツリーの正しい表示をサポート。 Support arrow-flight-sql protocol getStreamCatalogs, getStreamSchemas, getStreamTables #46217。

  2. 複数のBEがクラスター外部からアクセス可能な同一IPを共有する場合、クエリ結果を正しく転送してADBC Clientに返すことができる。 Arrow flight server supports data forwarding when BE uses public vip

  3. 複数エンドポイントの並列読み取りをサポート。 Arrow Flight support multiple endpoints

  4. クエリエラーFE not found arrow flight schemaを修正。 Fix FE not found arrow flight schema

  5. NULLを許可するカラムを読み取る際のエラーBooleanBuilder::AppendValuesを修正。 Fix Doris NULL column conversion to arrow batch

  6. show processlistで重複するConnection IDが表示される問題を修正。 Fix arrow-flight-sql ConnectContext to use a unified ID #46284

  7. DatetimeとDatetimeV2型を読み取る際にタイムゾーンが失われ、実際のデータより8時間少ないdatetimeになる問題を修正。 Fix time zone issues and accuracy issues #38215

v2.1.7​

  1. 頻繁なログ出力Connection wait_timeoutを修正。 Fix kill timeout FlightSqlConnection and FlightSqlConnectProcessor close

  2. Arrow Flight Bearer TokenがCacheから期限切れで削除される問題を修正。 Fix Arrow Flight bearer token cache evict after expired

v2.1.6​

  1. クエリエラー0.0.0.0:xxx, connection refusedを修正。 Fix return result from FE Arrow Flight server error 0.0.0.0:xxx, connection refused

  2. クエリエラーReach limit of connectionsを修正。 Fix exceed user property max connection cause Reach limit of connections #39127

以前のバージョンでは、SET PROPERTY FOR 'root' 'max_user_connections' = '1024';を実行して現在のユーザーの最大接続数を1024に変更することで、一時的に回避できます。

以前のバージョンではArrow Flight接続数をqe_max_connection/2未満に制限するのみで、qe_max_connectionはすべてのfeユーザーの総接続数でデフォルトは1024ですが、単一ユーザーのArrow Flight接続数をUserProperty内のmax_user_connections未満に制限せず、デフォルトは100です。そのためArrow Flight接続数が現在のユーザーの接続数上限を超えるとエラーReach limit of connectionsが報告されるため、現在のユーザーのmax_user_connectionsを増やす必要があります。

問題の詳細については以下を参照：Questions

  1. Conf arrow_flight_result_sink_buffer_size_rowsを追加して、1回で返されるクエリ結果のArrowBatchサイズの変更をサポート、デフォルトは4096 * 8。 Add config arrow_flight_result_sink_buffer_size_rows

v2.1.5​

  1. Arrow Flight SQLクエリ結果が空になる問題を修正。 Fix arrow flight result sink #36827

Doris v2.1.4では大量のデータを読み取る際にエラーが報告される可能性があります。詳細については以下を参照：Questions

3.0 リリースノート​

v3.0.5​

  1. DorisデータのArrowへのシリアライゼーション問題を修正。 Fix UT DataTypeSerDeArrowTest of Array/Map/Struct/Bitmap/HLL/Decimal256 types
  • Decimal256型の読み取りに失敗；
  • DatetimeV2型の読み取りで微細なエラー；
  • DateV2型の読み取りで結果が正しくない；
  • IPV4/IPV6型の読み取りで結果がNULLの場合のエラー；

v3.0.4​

  1. DBeaverなどのBIツールがarrow-flight-sqlプロトコルを使用してDorisに接続することをサポートし、メタデータツリーの正しい表示をサポート。 Support arrow-flight-sql protocol getStreamCatalogs, getStreamSchemas, getStreamTables #46217。

  2. 複数エンドポイントの並列読み取りをサポート。 Arrow Flight support multiple endpoints

  3. NULLを許可するカラムを読み取る際のエラーBooleanBuilder::AppendValuesを修正。 Fix Doris NULL column conversion to arrow batch

  4. show processlistで重複するConnection IDが表示される問題を修正。 Fix arrow-flight-sql ConnectContext to use a unified ID #46284

  5. Doris Arrow Flight SQLクエリが失敗して空の結果を返し、実際のエラー情報を返さない問題を修正。 Fix query result is empty and not return query error message

v3.0.3​

  1. クエリエラー0.0.0.0:xxx, connection refusedを修正。 Fix return result from FE Arrow Flight server error 0.0.0.0:xxx, connection refused

  2. クエリエラーReach limit of connectionsを修正。 Fix exceed user property max connection cause Reach limit of connections #39127

以前のバージョンでは、SET PROPERTY FOR 'root' 'max_user_connections' = '1024';を実行して現在のユーザーの最大接続数を1024に変更することで、一時的に回避できます。

以前のバージョンではArrow Flight接続数をqe_max_connection/2未満に制限するのみで、qe_max_connectionはすべてのfeユーザーの総接続数でデフォルトは1024ですが、単一ユーザーのArrow Flight接続数をUserProperty内のmax_user_connections未満に制限せず、デフォルトは100です。そのためArrow Flight接続数が現在のユーザーの接続数上限を超えるとエラーReach limit of connectionsが報告されるため、現在のユーザーのmax_user_connectionsを増やす必要があります。

問題の詳細については以下を参照：Questions

  1. 頻繁なログ出力Connection wait_timeoutを修正。 Fix kill timeout FlightSqlConnection and FlightSqlConnectProcessor close

  2. Arrow Flight Bearer TokenがCacheから期限切れで削除される問題を修正。 Fix Arrow Flight bearer token cache evict after expired

  3. 複数のBEがクラスター外部からアクセス可能な同一IPアドレスを共有する場合、クエリ結果を正しく転送してADBC Clientに返すことができる。 Arrow flight server supports data forwarding when BE uses public vip

  4. クエリエラーFE not found arrow flight schemaを修正。 Fix FE not found arrow flight schema

  5. DatetimeとDatetimeV2型を読み取る際にタイムゾーンが失われ、実際のデータより8時間少ないdatetimeになる問題を修正。 Fix time zone issues and accuracy issues #38215

v3.0.2​

  1. Conf arrow_flight_result_sink_buffer_size_rowsを追加して、単一トランザクションで返されるクエリ結果のArrowBatchサイズの変更をサポート、デフォルトは4096 * 8。 Add config arrow_flight_result_sink_buffer_size_rows

v3.0.1​

  1. クエリ結果が欠落し、クエリ結果行数 = 実際の行数 / BEの数 Fix get Schema failed when enable_parallel_result_sink is false #37779

Doris 3.0.0では、クエリの最外層が集約の場合、select k1, sum(k2) from xxx group by k1のようなSQLで（クエリ結果行数 = 実際の行数 / BEの数）に遭遇する可能性があります。これはsupport parallel result sinkによって導入された問題です。Fix get Schema failed when enable_parallel_result_sink is falseは一時的な修正で、Arrow Flight support multiple endpointsが複数エンドポイントの並列読み取りをサポートした後に正式に修正されます。