E-MapReduce:Routine Load

必須パラメーター

オプションパラメーター

データ整形

ジョブの動作

Kafka ソース設定

フロントエンド (FE) 構成 (fe.conf)

例：認証なしの Kafka から JSON データをロード

この例では、example_tbl2_ordertest という名前のロードジョブを作成し、ordertest2 トピックからメッセージを継続的に取得して、example_tbl テーブルにロードします。開始オフセットは、利用可能な最も古いオフセットからとします。pay_dt カラムは、pay_time の UNIX タイムスタンプを from_unixtime() 関数で変換して導出されます。

CREATE ROUTINE LOAD load_test.example_tbl2_ordertest ON example_tbl
COLUMNS(commodity_id, customer_name, country, pay_time, price, pay_dt=from_unixtime(pay_time, '%Y%m%d'))
PROPERTIES
(
    "desired_concurrent_number" = "5",
    "format" = "json",
    "jsonpaths" = "[\"$.commodity_id\",\"$.customer_name\",\"$.country\",\"$.pay_time\",\"$.price\"]"
)
FROM KAFKA
(
    "kafka_broker_list" = "<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest2",
    "kafka_partitions" = "0,1,2,3,4",
    "property.kafka_default_offsets" = "OFFSET_BEGINNING"
);

SHOW ROUTINE LOAD FOR load_test.example_tbl2_ordertest\G を実行し、State が RUNNING であり、loadedRows（Statistic 内）が増加していることを確認してください。

例：SSL を使用したデータロード

SSL 認証を有効化するには、以下のプロパティを追加します。

-- セキュリティプロトコル
"property.security.protocol" = "ssl",

-- CA 証明書のパス
"property.ssl.ca.location" = "FILE:ca-cert",

-- クライアント証明書およびキー（Kafka サーバーがクライアント認証を強制する場合に必要）
"property.ssl.certificate.location" = "FILE:client.pem",
"property.ssl.key.location" = "FILE:client.key",
"property.ssl.key.password" = "******"

CREATE FILE ステートメントを使用して証明書ファイルをアップロードする場合、Object Storage Service (OSS) の HTTP アドレスを URL として指定します。OSS のエンドポイントの一覧については、「IPv6 対応のエンドポイントを使用して OSS にアクセスする」をご参照ください。

CREATE FILE 文の構文については、CREATE FILE をご参照ください。

ジョブステータスの確認

データベース内のすべてのロードジョブ（停止およびキャンセル済みのジョブを含む）を表示するには：

USE load_test;
SHOW ALL ROUTINE LOAD;

特定の実行中のジョブを表示するには：

SHOW ROUTINE LOAD FOR load_test.example_tbl2_ordertest;

また、EMR StarRocks Manager コンソールでもロードジョブのステータスを確認できます。EMR StarRocks Manager ページに移動し、左側のナビゲーションウィンドウから メタデータ管理 をクリックし、目的のデータベース名をクリックした後、タスク タブをクリックします。

出力例：

*************************** 1. 行 ***************************

                  Id: 14093
                Name: routine_load_wikipedia
          CreateTime: 2020-05-16 16:00:48
           PauseTime: N/A
             EndTime: N/A
              DbName: default_cluster:load_test
           TableName: routine_wiki_edit
               State: RUNNING
      DataSourceType: KAFKA
      CurrentTaskNum: 1
       JobProperties: {"partitions":"*","columnToColumnExpr":"event_time,channel,user,is_anonymous,is_minor,is_new,is_robot,is_unpatrolled,delta,added,deleted","maxBatchIntervalS":"10","whereExpr":"*","maxBatchSizeBytes":"104857600","columnSeparator":"','","maxErrorNum":"1000","currentTaskConcurrentNum":"1","maxBatchRows":"200000"}
DataSourceProperties: {"topic":"starrocks-load","currentKafkaPartitions":"0","brokerList":"localhost:9092"}
    CustomProperties: {}
           Statistic: {"receivedBytes":150821770,"errorRows":122,"committedTaskNum":12,"loadedRows":2399878,"loadRowsRate":199000,"abortedTaskNum":1,"totalRows":2400000,"unselectedRows":0,"receivedBytesRate":12523000,"taskExecuteTimeMs":12043}
            Progress: {"0":"13634667"}
ReasonOfStateChanged:
        ErrorLogUrls: http://172.26.**.**:9122/api/_load_error_log?file=__shard_53/error_log_insert_stmt_47e8a1d107ed4932-8f1ddf7b01ad2fee_47e8a1d107ed4932_8f1ddf7b01ad2fee, http://172.26.**.**:9122/api/_load_error_log?file=__shard_54/error_log_insert_stmt_e0c0c6b040c044fd-a162b16f6bad53e6_e0c0c6b040c044fd_a162b16f6bad53e6, http://172.26.**.**:9122/api/_load_error_log?file=__shard_55/error_log_insert_stmt_ce4c95f0c72440ef-a442bb300bd743c8_ce4c95f0c72440ef_a442bb300bd743c8
            OtherMsg:
1 行が見つかりました (0.00 秒)

主要な健全性指標：

• State： RUNNING である必要があります。その他の状態については、下記のジョブ状態表をご参照ください。

• loadedRows および errorRows（Statistic 内）： errorRows が loadedRows に対して増加傾向にある場合、ジョブが自動的に一時停止する可能性があります。ReasonOfStateChanged および ErrorLogUrls 内の URL を確認してください。

• Progress： 各パーティションの現在の Kafka オフセットを示します。定期的に確認し、進捗が継続的に更新されていることを確認してください。

`Statistic` 出力のフィールド：

ジョブ状態

ジョブが自動的に PAUSED 状態に遷移した場合、エラー行数が max_error_number を超えたことが原因です。ReasonOfStateChanged および ErrorLogUrls 内のエラーログ URL を確認し、原因を特定してください。

ロードジョブの一時停止

PAUSE ROUTINE LOAD FOR <job_name>;

ジョブは PAUSED 状態になります。データインジェストは停止しますが、ジョブ自体は終了しません。Statistic および Progress の値の更新も停止します。RESUME ROUTINE LOAD を実行することで再開できます。

ロードジョブの再開

RESUME ROUTINE LOAD FOR <job_name>;

ジョブはまず NEED_SCHEDULE 状態になり、その後再スケジュールされて RUNNING 状態に遷移します。

ロードジョブの変更

PAUSED 状態のジョブのみ変更可能です。

ALTER ROUTINE LOAD FOR <job_name>
PROPERTIES
(
    "desired_concurrent_number" = "6"
);

ロードジョブの停止

STOP ROUTINE LOAD FOR <job_name>;

ジョブは STOPPED 状態になり、完全に終了します。Statistic および Progress の更新は停止し、SHOW ROUTINE LOAD の出力にも表示されなくなります。データインジェストを再開するには、新しいロードジョブを作成してください。

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

1. トピックを作成します。

   kafka-topics.sh --create \
     --topic order_sr_topic \
     --replication-factor 3 \
     --partitions 10 \
     --bootstrap-server core-1-1:9092,core-1-2:9092,core-1-3:9092

2. コンソールプロデューサーを起動します。

   kafka-console-producer.sh \
     --broker-list core-1-1:9092 \
     --topic order_sr_topic

3. 以下の各行を入力します。

   2020050802,2020-05-08,Johann Georg Faust,Deutschland,male,895
   2020050802,2020-05-08,Julien Sorel,France,male,893
   2020050803,2020-05-08,Dorian Grey,UK,male,1262
   2020051001,2020-05-10,Tess Durbeyfield,US,female,986
   2020051101,2020-05-11,Edogawa Conan,japan,male,8924

   各行は 6 つのカラムで構成されています：order_id、pay_dt、customer_name、nationality、gender（スキップ対象）、price。

ステップ 2：送信先テーブルの作成

カラム 1、2、3、4、6 に対応する StarRocks テーブルを作成し、gender カラムはスキップします。

CREATE TABLE load_test.routine_load_tbl_csv (
    `order_id`       bigint      NOT NULL COMMENT "注文 ID",
    `pay_dt`         date        NOT NULL COMMENT "購入日付",
    `customer_name`  varchar(26) NULL     COMMENT "顧客名",
    `nationality`    varchar(26) NULL     COMMENT "国",
    `price`          double      NULL     COMMENT "支払金額"
)
ENGINE = OLAP
PRIMARY KEY (order_id, pay_dt)
DISTRIBUTED BY HASH(`order_id`) BUCKETS 5;

ステップ 3：ロードジョブの作成

COLUMNS 句では、6 つのソースカラムすべてをマップし、5 番目のカラムをスキップするためにプレースホルダー（temp_gender）を使用します。

CREATE ROUTINE LOAD load_test.routine_load_tbl_ordertest_csv ON routine_load_tbl_csv
COLUMNS TERMINATED BY ",",
COLUMNS (order_id, pay_dt, customer_name, nationality, temp_gender, price)
PROPERTIES
(
    "desired_concurrent_number" = "5"
)
FROM KAFKA
(
    "kafka_broker_list" = "192.168.**.**:9092,192.168.**.**:9092,192.168.**.**:9092",
    "kafka_topic" = "order_sr_topic",
    "kafka_partitions" = "0,1,2,3,4",
    "property.kafka_default_offsets" = "OFFSET_BEGINNING"
);

ステップ 4：ジョブの検証

SHOW ROUTINE LOAD FOR routine_load_tbl_ordertest_csv;

以下の 3 つのフィールドを確認してください。

1. State の状態は RUNNING です。

2. DataSourceProperties には、"topic":"order_sr_topic" および正しいブローカーアドレスが表示されます。

3. Statistic は、loadedRows が増加しており、errorRows が 0 になっていることを示しています。

送信先テーブルをクエリして、データが正しくロードされたことを確認します。

SELECT * FROM load_test.routine_load_tbl_csv LIMIT 5;

期待される結果：