E-MapReduce:Routine Load

CSV データのインポート

ステップ 1：データセットの準備

Kafka のトピック my_topic に次の形式の CSV メッセージが含まれていると仮定します。

val_k1,val_k2,val_k3,val_v1,val_v2,val_v3_raw

この 6 つのフィールドは、列 k1、k2、k3、v1、v2、および v3 を導出するために使用される生値（raw value）に対応します。

ステップ 2：Routine Load ジョブの作成

CREATE ROUTINE LOAD example_db.test1 ON example_tbl
        COLUMNS TERMINATED BY ",",
        COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100)
        PROPERTIES
        (
            "desired_concurrent_number"="3",
            "max_batch_interval" = "20",
            "max_batch_rows" = "300000",
            "max_batch_size" = "209715200",
            "strict_mode" = "false"
        )
        FROM KAFKA
        (
            "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
            "kafka_topic" = "my_topic",
            "property.group.id" = "xxx",
            "property.client.id" = "xxx",
            "property.kafka_default_offsets" = "OFFSET_BEGINNING"
        );

列マッピング： Doris は、CSV フィールドを位置順に COLUMNS リストにマップし、その後、リスト内の列名をテーブルスキーマの列名と照合してマッピングします。v3 = k1 * 100 という式は変換処理であり、Doris は v3 をソースから直接読み取るのではなく、k1 から計算して導出します。ソースのフィールド名、数、順序がテーブルの列と完全に一致する場合は、COLUMNS 句を省略できます。

OFFSET_BEGINNING は、利用可能な最も古いメッセージから消費を開始します。最新のメッセージから開始する場合は、代わりに OFFSET_END を使用してください。

Strict モードでの CSV データのインポート

Strict モードでは、型変換に失敗した行（例：整数にキャストできない文字列）がフィルターで除外されます。データ品質が特に重要な場合に使用してください。

CREATE ROUTINE LOAD example_db.test1 ON example_tbl
        COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100),
        WHERE k1 > 100 and k2 like "%doris%"
        PROPERTIES
        (
            "desired_concurrent_number"="3",
            "max_batch_interval" = "20",
            "max_batch_rows" = "300000",
            "max_batch_size" = "209715200",
            "strict_mode" = "true"
        )
        FROM KAFKA
        (
            "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
            "kafka_topic" = "my_topic",
            "kafka_partitions" = "0,1,2,3",
            "kafka_offsets" = "101,0,0,200"
        );

kafka_partitions および kafka_offsets は、各パーティションの開始オフセットを指定します。両パラメーターのエントリー数は一致している必要があります。

Strict モードの動作

以下の表は、Strict モードが型変換失敗時の行に与える影響を示しています。

*TinyInt 列（NULL 許容）：*

*Decimal(1,0) 列（NULL 許容）：*

値 10 は Decimal 型として有効な値であるため Strict モードを通過しますが、Decimal(1,0) の精度を超えるため、抽出・変換・ロード（ETL）フェーズでフィルターされます。

JSON データのインポート

Routine Load では、2 種類の JSON レイアウトをサポートしています。

単一の JSON オブジェクト（メッセージごとに 1 レコード）：

{"category":"a9jadhx","author":"test","price":895}

JSON 配列（メッセージごとに複数のレコード）：

[
    {"category":"11","author":"4avc","price":895,"timestamp":1589191587},
    {"category":"22","author":"2avc","price":895,"timestamp":1589191487},
    {"category":"33","author":"3avc","price":342,"timestamp":1589191387}
]

ステップ 1：対象テーブルの作成

CREATE TABLE `example_tbl` (
   `category` varchar(24) NULL COMMENT "",
   `author` varchar(24) NULL COMMENT "",
   `timestamp` bigint(20) NULL COMMENT "",
   `dt` int(11) NULL COMMENT "",
   `price` double REPLACE
) ENGINE=OLAP
AGGREGATE KEY(`category`,`author`,`timestamp`,`dt`)
COMMENT "OLAP"
PARTITION BY RANGE(`dt`)
(
  PARTITION p0 VALUES [("-2147483648"), ("20200509")),
    PARTITION p20200509 VALUES [("20200509"), ("20200510")),
    PARTITION p20200510 VALUES [("20200510"), ("20200511")),
    PARTITION p20200511 VALUES [("20200511"), ("20200512"))
)
DISTRIBUTED BY HASH(`category`,`author`,`timestamp`) BUCKETS 4
PROPERTIES (
    "replication_num" = "1"
);

ステップ 2a：単一オブジェクト JSON のインポート

CREATE ROUTINE LOAD example_db.test_json_label_1 ON table1
COLUMNS(category,price,author)
PROPERTIES
(
    "desired_concurrent_number"="3",
    "max_batch_interval" = "20",
    "max_batch_rows" = "300000",
    "max_batch_size" = "209715200",
    "strict_mode" = "false",
    "format" = "json"
)
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
    "kafka_topic" = "my_topic",
    "kafka_partitions" = "0,1,2",
    "kafka_offsets" = "0,0,0"
 );

ステップ 2b：JSON 配列のインポート

CREATE ROUTINE LOAD example_db.test1 ON example_tbl
COLUMNS(category, author, price, timestamp, dt=from_unixtime(timestamp, '%Y%m%d'))
PROPERTIES
(
    "desired_concurrent_number"="3",
    "max_batch_interval" = "20",
    "max_batch_rows" = "300000",
    "max_batch_size" = "209715200",
    "strict_mode" = "false",
    "format" = "json",
    "jsonpaths" = "[\"$.category\",\"$.author\",\"$.price\",\"$.timestamp\"]",
    "strip_outer_array" = "true"
)
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
    "kafka_topic" = "my_topic",
    "kafka_partitions" = "0,1,2",
    "kafka_offsets" = "0,0,0"
);

メッセージに JSON 配列が含まれる場合は、strip_outer_array を true に設定します。jsonpaths を使用して、特定の JSON フィールドをテーブルの列にマップします。dt 列はソースデータには存在しないため、式 dt=from_unixtime(timestamp, '%Y%m%d') を用いて timestamp から導出します。

SSL を使用した Kafka クラスターへの接続

Kafka クラスターが SSL 認証を要求する場合は、まず証明書ファイルをアップロードし、その後ジョブ内で参照します。

ステップ 1：証明書ファイルのアップロード

CREATE FILE "ca.pem" PROPERTIES("url" = "https://example_url/kafka-key/ca.pem", "catalog" = "kafka");
CREATE FILE "client.key" PROPERTIES("url" = "https://example_url/kafka-key/client.key", "catalog" = "kafka");
CREATE FILE "client.pem" PROPERTIES("url" = "https://example_url/kafka-key/client.pem", "catalog" = "kafka");

すべての SSL 接続に対して、CA 証明書（ca.pem）をアップロードしてください。Kafka クラスターがクライアント認証も要求する場合は、client.pem および client.key も併せてアップロードしてください。CREATE FILE コマンドの詳細については、HELP CREATE FILE; を実行してください。

ステップ 2：SSL を使用した Routine Load ジョブの作成

CREATE ROUTINE LOAD db1.job1 on tbl1
PROPERTIES
(
    "desired_concurrent_number"="1"
)
FROM KAFKA
(
    "kafka_broker_list"= "broker1:9091,broker2:9091",
    "kafka_topic" = "my_topic",
    "property.security.protocol" = "ssl",
    "property.ssl.ca.location" = "FILE:ca.pem",
    "property.ssl.certificate.location" = "FILE:client.pem",
    "property.ssl.key.location" = "FILE:client.key",
    "property.ssl.key.password" = "abcd***"
);

Doris は C++ クライアントライブラリ librdkafka を使用して Kafka に接続します。サポートされる接続プロパティの完全な一覧については、librdkafka 構成リファレンスをご参照ください。

ジョブのステータス確認

SHOW ROUTINE LOAD;
SHOW ROUTINE LOAD TASK;

SHOW ROUTINE LOAD は、ジョブの状態および統計情報を返します。SHOW ROUTINE LOAD TASK は、ジョブ内の個々のタスクのステータスを表示します。いずれのコマンドも、RUNNING 状態のジョブのみを表示します。完了済みおよび未開始のジョブは表示されません。

構文の詳細については、HELP SHOW ROUTINE LOAD; または HELP SHOW ROUTINE LOAD TASK; を実行してください。

ジョブの変更

ALTER ROUTINE LOAD ...;

ALTER ROUTINE LOAD を使用して、既に作成されたジョブのプロパティを更新します。構文の詳細については、HELP ALTER ROUTINE LOAD; を実行してください。

ジョブの一時停止、再開、停止

構文の詳細については、HELP PAUSE ROUTINE LOAD;、HELP RESUME ROUTINE LOAD;、または HELP STOP ROUTINE LOAD; を実行してください。

一時停止中のジョブが自動的に回復する場合

Doris は、一時的なエラー（例：一時的な BE の障害）によって一時停止されたジョブについて、自動回復を試行します。period_of_auto_resume_min（デフォルト：5 分）の期間内に最大 3 回再試行します。すべての再試行が失敗した場合、ジョブはロックされ、手動による RESUME が必要になります。

スキーマ変更とインポート

Routine Load ジョブは、SCHEMA CHANGE および ROLLUP 操作をブロックしません。ただし、SCHEMA CHANGE によってソースデータとターゲットテーブルの列が不一致になると、エラー行数が増加し、最終的にジョブが一時停止します。これを防ぐには、ジョブ内で明示的な列マッピングを指定し、NULL 許容列または DEFAULT 値を持つ列を使用してください。

パーティションの削除

Routine Load ジョブがインポート中であるテーブルパーティションを削除すると、パーティションが見つからないためジョブが一時停止します。

同時実行操作

Routine Load ジョブは、LOAD および INSERT 操作と競合することなく同時に実行されます。テーブルに対して DELETE を実行する前に、Routine Load ジョブを一時停止し、割り当てられたすべてのタスクが完了するのを待ってから、DELETE を実行してください。

データベースおよびテーブルの削除

ターゲットのデータベースまたはテーブルを削除すると、Routine Load ジョブは直ちにキャンセルされます。

Kafka トピックの自動作成

ジョブで指定された Kafka トピックが存在しない場合、Kafka ブローカーは auto.create.topics.enable の設定に基づいて、自動的にトピックを作成できます。

• auto.create.topics.enable=true：Kafka は num.partitions で定義されたパーティション数でトピックを作成します。ジョブは通常どおり消費を開始します。

• auto.create.topics.enable=false：Kafka はトピックを作成しません。ジョブは、トピックが存在し、データが格納されるまで一時停止します。

トピックが確実に自動作成されるようにするには、クラスター内のすべてのブローカーで auto.create.topics.enable=true を設定してください。

ネットワーク隔離

CIDR ベースの隔離や制限付き DNS を使用する環境では、以下の点に注意してください。

• Doris は、kafka_broker_list 内のすべてのブローカーに到達可能である必要があります。

• Kafka クラスターで advertised.listeners が構成されている場合、Doris はブローカー一覧のアドレスだけでなく、これらのリスナーのアドレスにも到達可能である必要があります。