Routine Load

Routine Load を使用したデータロード

ヒント

Try Routine Load out in this Quick Start

このトピックでは、Kafka メッセージ（イベント）を StarRocks にストリームするための Routine Load ジョブの作成方法を紹介し、Routine Load に関する基本的な概念を説明します。

ストリームのメッセージを StarRocks に継続的にロードするには、Kafka トピックにメッセージストリームを保存し、Routine Load ジョブを作成してメッセージを消費します。Routine Load ジョブは StarRocks に永続化され、トピック内のすべてまたは一部のパーティションのメッセージを消費するための一連のロードタスクを生成し、メッセージを StarRocks にロードします。

Routine Load ジョブは、StarRocks にロードされたデータが失われたり重複したりしないことを保証するために、正確に一度だけの配信セマンティクスをサポートしています。

Routine Load はデータロード時のデータ変換をサポートし、データロード中に UPSERT および DELETE 操作によるデータ変更をサポートします。詳細については、Transform data at loading および Change data through loading を参照してください。

You can load data into StarRocks tables only as a user who has the INSERT privilege on those StarRocks tables. If you do not have the INSERT privilege, follow the instructions provided in GRANT to grant the INSERT privilege to the user that you use to connect to your StarRocks cluster. The syntax is GRANT INSERT ON TABLE <table_name> IN DATABASE <database_name> TO { ROLE <role_name> | USER <user_identity>}.

サポートされているデータ形式

Routine Load は現在、Kafka クラスターからの CSV、JSON、および Avro（v3.0.1 以降でサポート）形式のデータの消費をサポートしています。

注意

CSV データについては、以下の点に注意してください:

• テキスト区切り文字として、長さが 50 バイトを超えない UTF-8 文字列（カンマ（,）、タブ、パイプ（|）など）を使用できます。
• Null 値は \N を使用して示されます。たとえば、データファイルが 3 列で構成され、そのデータファイルのレコードが最初と最後の列にデータを持ち、2 番目の列にデータがない場合、この状況では 2 番目の列に \N を使用して Null 値を示す必要があります。つまり、レコードは a,\N,b としてコンパイルされる必要があります。a,,b は、レコードの 2 番目の列が空の文字列を持っていることを示します。

基本概念

用語

• Load job

  Routine Load ジョブは長時間実行されるジョブです。そのステータスが RUNNING である限り、ロードジョブは Kafka クラスターのトピック内のメッセージを消費し、データを StarRocks にロードするための 1 つまたは複数の同時ロードタスクを継続的に生成します。

• Load task

  ロードジョブは特定のルールに基づいて複数のロードタスクに分割されます。ロードタスクはデータロードの基本単位です。個々のイベントとして、ロードタスクは Stream Load に基づくロードメカニズムを実装します。複数のロードタスクが異なるトピックのパーティションからメッセージを同時に消費し、データを StarRocks にロードします。

ワークフロー

1. Routine Load ジョブを作成します。 Kafka からデータをロードするには、CREATE ROUTINE LOAD ステートメントを実行して Routine Load ジョブを作成する必要があります。FE はステートメントを解析し、指定したプロパティに従ってジョブを作成します。

2. FE はジョブを複数のロードタスクに分割します。

   FE は特定のルールに基づいてジョブを複数のロードタスクに分割します。各ロードタスクは個別のトランザクションです。 分割ルールは次のとおりです:

   • FE は、Kafka トピックのパーティション数と生存している BE ノードの数に基づいて、ロードタスクの実際の同時数を計算します。
   • FE は計算された実際の同時数に基づいてジョブをロードタスクに分割し、タスクをタスクキューに配置します。

   各 Kafka トピックは複数のパーティションで構成されています。トピックパーティションとロードタスクの関係は次のとおりです:

   • パーティションはロードタスクに一意に割り当てられ、そのパーティションからのすべてのメッセージはロードタスクによって消費されます。
   • ロードタスクは 1 つ以上のパーティションからメッセージを消費できます。
   • すべてのパーティションはロードタスク間で均等に分配されます。

3. 複数のロードタスクが同時に複数の Kafka トピックパーティションからメッセージを消費し、データを StarRocks にロードします。

   1. FE はロードタスクをスケジュールして提出します: FE はキュー内のロードタスクをタイムリーにスケジュールし、選択された Coordinator BE ノードに割り当てます。ロードタスク間の間隔は、構成項目 max_batch_interval によって定義されます。FE はロードタスクをすべての BE ノードに均等に分配します。max_batch_interval の詳細については、CREATE ROUTINE LOAD を参照してください。

   2. Coordinator BE はロードタスクを開始し、パーティション内のメッセージを消費し、データを解析してフィルタリングします。ロードタスクは、事前に定義されたメッセージの量が消費されるか、事前に定義された時間制限に達するまで続きます。メッセージバッチサイズと時間制限は、FE 構成の max_routine_load_batch_size と routine_load_task_consume_second に定義されています。詳細については、FE Configuration を参照してください。その後、Coordinator BE はメッセージを Executor BEs に分配します。Executor BEs はメッセージをディスクに書き込みます。

      注意

      StarRocks は、SASL_SSL、SAS_PLAINTEXT、SSL、および PLAINTEXT を含むセキュリティプロトコルを介した Kafka へのアクセスをサポートしています。このトピックでは、PLAINTEXT を介した Kafka への接続を例として使用しています。他のセキュリティプロトコルを介して Kafka に接続する必要がある場合は、CREATE ROUTINE LOAD を参照してください。

4. FE はデータを継続的にロードするための新しいロードタスクを生成します。 Executor BEs がデータをディスクに書き込んだ後、Coordinator BE はロードタスクの結果を FE に報告します。その結果に基づいて、FE はデータを継続的にロードするための新しいロードタスクを生成します。または、FE は失敗したタスクを再試行して、StarRocks にロードされたデータが失われたり重複したりしないことを確認します。

Routine Load ジョブを作成する

次の 3 つの例では、Kafka で CSV 形式、JSON 形式、および Avro 形式のデータを消費し、Routine Load ジョブを作成して StarRocks にデータをロードする方法を説明します。詳細な構文とパラメータの説明については、CREATE ROUTINE LOAD を参照してください。

CSV 形式のデータをロードする

このセクションでは、Kafka クラスターで CSV 形式のデータを消費し、StarRocks にデータをロードするための Routine Load ジョブを作成する方法を説明します。

データセットを準備する

Kafka クラスターのトピック ordertest1 に CSV 形式のデータセットがあるとします。データセット内の各メッセージには、注文 ID、支払日、顧客名、国籍、性別、価格の 6 つのフィールドが含まれています。

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
2020050901,2020-05-09,Anna Karenina",Russia,female,175
2020051001,2020-05-10,Tess Durbeyfield,US,female,986
2020051101,2020-05-11,Edogawa Conan,japan,male,8924

テーブルを作成する

CSV 形式のデータのフィールドに基づいて、データベース example_db にテーブル example_tbl1 を作成します。次の例では、CSV 形式のデータの顧客性別フィールドを除く 5 つのフィールドを持つテーブルを作成します。

CREATE TABLE example_db.example_tbl1 ( 
    `order_id` bigint NOT NULL COMMENT "Order ID",
    `pay_dt` date NOT NULL COMMENT "Payment date", 
    `customer_name` varchar(26) NULL COMMENT "Customer name", 
    `nationality` varchar(26) NULL COMMENT "Nationality", 
    `price`double NULL COMMENT "Price"
) 
ENGINE=OLAP 
DUPLICATE KEY (order_id,pay_dt) 
DISTRIBUTED BY HASH(`order_id`); 

注意

v2.5.7 以降、StarRocks はテーブルを作成する際やパーティションを追加する際に、バケット数（BUCKETS）を自動的に設定できます。バケット数を手動で設定する必要はありません。詳細については、set the number of buckets を参照してください。

Routine Load ジョブを提出する

次のステートメントを実行して、トピック ordertest1 のメッセージを消費し、テーブル example_tbl1 にデータをロードする Routine Load ジョブ example_tbl1_ordertest1 を提出します。ロードタスクは、指定されたトピックのパーティションの初期オフセットからメッセージを消費します。

CREATE ROUTINE LOAD example_db.example_tbl1_ordertest1 ON example_tbl1
COLUMNS TERMINATED BY ",",
COLUMNS (order_id, pay_dt, customer_name, nationality, temp_gender, price)
PROPERTIES
(
    "desired_concurrent_number" = "5"
)
FROM KAFKA
(
    "kafka_broker_list" = "<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest1",
    "kafka_partitions" = "0,1,2,3,4",
    "property.kafka_default_offsets" = "OFFSET_BEGINNING"
);

ロードジョブを提出した後、SHOW ROUTINE LOAD ステートメントを実行して、ロードジョブのステータスを確認できます。

• ロードジョブ名

  1 つのテーブルに複数のロードジョブが存在する可能性があります。そのため、ロードジョブを対応する Kafka トピックとロードジョブが提出された時間で名前を付けることをお勧めします。これにより、各テーブルのロードジョブを区別するのに役立ちます。

• カラムセパレータ

  プロパティ COLUMN TERMINATED BY は、CSV 形式のデータのカラムセパレータを定義します。デフォルトは \t です。

• Kafka トピックパーティションとオフセット

  プロパティ kafka_partitions と kafka_offsets を指定して、メッセージを消費するパーティションとオフセットを指定できます。たとえば、ロードジョブがトピック ordertest1 の Kafka パーティション "0,1,2,3,4" から初期オフセットでメッセージを消費するようにしたい場合、プロパティを次のように指定できます。ロードジョブが Kafka パーティション "0,1,2,3,4" からメッセージを消費し、各パーティションに対して個別の開始オフセットを指定する必要がある場合、次のように設定できます。

  "kafka_partitions" ="0,1,2,3,4",
  "kafka_offsets" = "OFFSET_BEGINNING, OFFSET_END, 1000, 2000, 3000"

  プロパティ property.kafka_default_offsets を使用して、すべてのパーティションのデフォルトオフセットを設定することもできます。

  "kafka_partitions" ="0,1,2,3,4",
  "property.kafka_default_offsets" = "OFFSET_BEGINNING"

  詳細については、CREATE ROUTINE LOAD を参照してください。

• データマッピングと変換

  CSV 形式のデータと StarRocks テーブルの間のマッピングと変換関係を指定するには、COLUMNS パラメータを使用する必要があります。

  データマッピング:

  • StarRocks は CSV 形式のデータのカラムを抽出し、それを COLUMNS パラメータで宣言されたフィールドに順番にマッピングします。

  • StarRocks は COLUMNS パラメータで宣言されたフィールドを抽出し、それを StarRocks テーブルのカラムに名前でマッピングします。

  データ変換:

  例では、CSV 形式のデータから顧客性別のカラムを除外しているため、COLUMNS パラメータのフィールド temp_gender はこのフィールドのプレースホルダーとして使用されます。他のフィールドは、StarRocks テーブル example_tbl1 のカラムに直接マッピングされます。

  データ変換の詳細については、Transform data at loading を参照してください。

  注意

  CSV 形式のデータのカラムの名前、数、および順序が StarRocks テーブルのそれと完全に一致する場合、COLUMNS パラメータを指定する必要はありません。

• タスクの同時実行性

  Kafka トピックパーティションが多く、BE ノードが十分にある場合、タスクの同時実行性を高めることでロードを加速できます。

  実際のロードタスクの同時実行性を高めるには、Routine Load ジョブを作成する際に、希望するロードタスクの同時実行性 desired_concurrent_number を増やすことができます。また、FE の動的構成項目 max_routine_load_task_concurrent_num（デフォルトの最大ロードタスクの同時実行性）を大きな値に設定することもできます。max_routine_load_task_concurrent_num の詳細については、FE configuration items を参照してください。

  実際のタスクの同時実行性は、生存している BE ノードの数、事前に指定された Kafka トピックパーティションの数、および desired_concurrent_number と max_routine_load_task_concurrent_num の値の最小値によって定義されます。

  例では、生存している BE ノードの数は 5、事前に指定された Kafka トピックパーティションの数は 5、max_routine_load_task_concurrent_num の値は 5 です。実際のロードタスクの同時実行性を高めるには、デフォルト値 3 から desired_concurrent_number を 5 に増やすことができます。

  プロパティの詳細については、CREATE ROUTINE LOAD を参照してください。

JSON 形式のデータをロードする

このセクションでは、Kafka クラスターで JSON 形式のデータを消費し、StarRocks にデータをロードするための Routine Load ジョブを作成する方法を説明します。

データセットを準備する

Kafka クラスターのトピック ordertest2 に JSON 形式のデータセットがあるとします。データセットには、商品 ID、顧客名、国籍、支払い時間、価格の 6 つのキーが含まれています。さらに、支払い時間のカラムを DATE 型に変換し、StarRocks テーブルの pay_dt カラムにロードしたいとします。

{"commodity_id": "1", "customer_name": "Mark Twain", "country": "US","pay_time": 1589191487,"price": 875}
{"commodity_id": "2", "customer_name": "Oscar Wilde", "country": "UK","pay_time": 1589191487,"price": 895}
{"commodity_id": "3", "customer_name": "Antoine de Saint-Exupéry","country": "France","pay_time": 1589191487,"price": 895}

注意 各 JSON オブジェクトは 1 行に 1 つの Kafka メッセージでなければなりません。そうでない場合、JSON 解析エラーが返されます。

テーブルを作成する

JSON 形式のデータのキーに基づいて、データベース example_db にテーブル example_tbl2 を作成します。

CREATE TABLE `example_tbl2` ( 
    `commodity_id` varchar(26) NULL COMMENT "Commodity ID", 
    `customer_name` varchar(26) NULL COMMENT "Customer name", 
    `country` varchar(26) NULL COMMENT "Country", 
    `pay_time` bigint(20) NULL COMMENT "Payment time", 
    `pay_dt` date NULL COMMENT "Payment date", 
    `price`double SUM NULL COMMENT "Price"
) 
ENGINE=OLAP
AGGREGATE KEY(`commodity_id`,`customer_name`,`country`,`pay_time`,`pay_dt`) 
DISTRIBUTED BY HASH(`commodity_id`); 

注意

v2.5.7 以降、StarRocks はテーブルを作成する際やパーティションを追加する際に、バケット数（BUCKETS）を自動的に設定できます。バケット数を手動で設定する必要はありません。詳細については、set the number of buckets を参照してください。

Routine Load ジョブを提出する

次のステートメントを実行して、トピック ordertest2 のメッセージを消費し、テーブル example_tbl2 にデータをロードする Routine Load ジョブ example_tbl2_ordertest2 を提出します。ロードタスクは、指定されたトピックのパーティションの初期オフセットからメッセージを消費します。

CREATE ROUTINE LOAD example_db.example_tbl2_ordertest2 ON example_tbl2
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 ステートメントを実行して、ロードジョブのステータスを確認できます。

• データ形式

  データ形式が JSON であることを定義するために、PROPERTIES 句で "format" = "json" を指定する必要があります。

• データマッピングと変換

  JSON 形式のデータと StarRocks テーブルの間のマッピングと変換関係を指定するには、COLUMNS パラメータとプロパティ jsonpaths を指定する必要があります。COLUMNS パラメータで指定されたフィールドの順序は JSON 形式のデータの順序と一致し、フィールドの名前は StarRocks テーブルの名前と一致する必要があります。プロパティ jsonpaths は、JSON データから必要なフィールドを抽出するために使用されます。これらのフィールドは COLUMNS プロパティによって名前が付けられます。

  例では、支払い時間フィールドを DATE データ型に変換し、StarRocks テーブルの pay_dt カラムにデータをロードする必要があるため、COLUMNS パラメータで from_unixtime 関数を使用する必要があります。他のフィールドは、テーブル example_tbl2 のフィールドに直接マッピングされます。

  データマッピング:

  • StarRocks は JSON 形式のデータの name と code キーを抽出し、それを jsonpaths プロパティで宣言されたキーにマッピングします。

  • StarRocks は jsonpaths プロパティで宣言されたキーを抽出し、それを COLUMNS パラメータで宣言されたフィールドに順番にマッピングします。

  • StarRocks は COLUMNS パラメータで宣言されたフィールドを抽出し、それを StarRocks テーブルのカラムに名前でマッピングします。

  データ変換:

  • 例では、キー pay_time を DATE データ型に変換し、StarRocks テーブルの pay_dt カラムにデータをロードする必要があるため、COLUMNS パラメータで from_unixtime 関数を使用する必要があります。他のフィールドは、テーブル example_tbl2 のフィールドに直接マッピングされます。

  • 例では、JSON 形式のデータから顧客性別のカラムを除外しているため、COLUMNS パラメータのフィールド temp_gender はこのフィールドのプレースホルダーとして使用されます。他のフィールドは、StarRocks テーブル example_tbl1 のカラムに直接マッピングされます。

    データ変換の詳細については、Transform data at loading を参照してください。

    注意

    JSON オブジェクトのキーの名前と数が StarRocks テーブルのフィールドと完全に一致する場合、COLUMNS パラメータを指定する必要はありません。

Avro 形式のデータをロードする

v3.0.1 以降、StarRocks は Routine Load を使用して Avro データのロードをサポートしています。

データセットを準備する

Avro スキーマ

1. 次の Avro スキーマファイル avro_schema.avsc を作成します:

   {
       "type": "record",
       "name": "sensor_log",
       "fields" : [
           {"name": "id", "type": "long"},
           {"name": "name", "type": "string"},
           {"name": "checked", "type" : "boolean"},
           {"name": "data", "type": "double"},
           {"name": "sensor_type", "type": {"type": "enum", "name": "sensor_type_enum", "symbols" : ["TEMPERATURE", "HUMIDITY", "AIR-PRESSURE"]}}  
       ]
   }

2. Avro スキーマを Schema Registry に登録します。

Avro データ

Avro データを準備し、Kafka トピック topic_0 に送信します。

テーブルを作成する

Avro データのフィールドに基づいて、StarRocks クラスターのターゲットデータベース example_db にテーブル sensor_log を作成します。テーブルのカラム名は Avro データのフィールド名と一致している必要があります。テーブルカラムと Avro データフィールド間のデータ型マッピングについては、[Data types mapping](#Data types mapping) を参照してください。

CREATE TABLE example_db.sensor_log ( 
    `id` bigint NOT NULL COMMENT "sensor id",
    `name` varchar(26) NOT NULL COMMENT "sensor name", 
    `checked` boolean NOT NULL COMMENT "checked", 
    `data` double NULL COMMENT "sensor data", 
    `sensor_type` varchar(26) NOT NULL COMMENT "sensor type"
) 
ENGINE=OLAP 
DUPLICATE KEY (id) 
DISTRIBUTED BY HASH(`id`); 

注意

v2.5.7 以降、StarRocks はテーブルを作成する際やパーティションを追加する際に、バケット数（BUCKETS）を自動的に設定できます。バケット数を手動で設定する必要はありません。詳細については、set the number of buckets を参照してください。

Routine Load ジョブを提出する

次のステートメントを実行して、Kafka トピック topic_0 の Avro メッセージを消費し、データベース sensor のテーブル sensor_log にデータをロードする Routine Load ジョブ sensor_log_load_job を提出します。ロードジョブは、指定されたトピックのパーティションの初期オフセットからメッセージを消費します。

CREATE ROUTINE LOAD example_db.sensor_log_load_job ON sensor_log  
PROPERTIES  
(  
    "format" = "avro"  
)  
FROM KAFKA  
(  
    "kafka_broker_list" = "<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>,...",
    "confluent.schema.registry.url" = "http://172.xx.xxx.xxx:8081",  
    "kafka_topic" = "topic_0",  
    "kafka_partitions" = "0,1,2,3,4,5",  
    "property.kafka_default_offsets" = "OFFSET_BEGINNING"  
);

• データ形式

  データ形式が Avro であることを定義するために、PROPERTIES 句で "format = "avro" を指定する必要があります。

• スキーマレジストリ

  Avro スキーマが登録されているスキーマレジストリの URL を指定するために、confluent.schema.registry.url を設定する必要があります。StarRocks はこの URL を使用して Avro スキーマを取得します。フォーマットは次のとおりです:

  confluent.schema.registry.url = http[s]://[<schema-registry-api-key>:<schema-registry-api-secret>@]<hostname|ip address>[:<port>]

• データマッピングと変換

  Avro 形式のデータと StarRocks テーブルの間のマッピングと変換関係を指定するには、COLUMNS パラメータとプロパティ jsonpaths を指定する必要があります。COLUMNS パラメータで指定されたフィールドの順序は jsonpaths プロパティのフィールドの順序と一致し、フィールドの名前は StarRocks テーブルの名前と一致する必要があります。プロパティ jsonpaths は、Avro データから必要なフィールドを抽出するために使用されます。これらのフィールドは COLUMNS プロパティによって名前が付けられます。

  データ変換の詳細については、Transform data at loading を参照してください。

  注意

  Avro レコードのフィールドの名前と数が StarRocks テーブルのカラムと完全に一致する場合、COLUMNS パラメータを指定する必要はありません。

ロードジョブを提出した後、SHOW ROUTINE LOAD ステートメントを実行して、ロードジョブのステータスを確認できます。

データ型のマッピング

ロードしたい Avro データフィールドと StarRocks テーブルカラム間のデータ型マッピングは次のとおりです:

プリミティブ型

Avro	StarRocks
nul	NULL
boolean	BOOLEAN
int	INT
long	BIGINT
float	FLOAT
double	DOUBLE
bytes	STRING
string	STRING

複合型

Avro	StarRocks
record	RECORD 全体またはそのサブフィールドを JSON として StarRocks にロードします。
enums	STRING
arrays	ARRAY
maps	JSON
union(T, null)	NULLABLE(T)
fixed	STRING

制限事項

• 現在、StarRocks はスキーマの進化をサポートしていません。
• 各 Kafka メッセージには単一の Avro データレコードのみを含める必要があります。

ロードジョブとタスクを確認する

ロードジョブを確認する

SHOW ROUTINE LOAD ステートメントを実行して、ロードジョブ example_tbl2_ordertest2 のステータスを確認します。StarRocks は、実行状態 State、統計情報（消費された総行数とロードされた総行数を含む）Statistics、およびロードジョブの進行状況 progress を返します。

ロードジョブの状態が自動的に PAUSED に変更された場合、エラーロウの数がしきい値を超えた可能性があります。このしきい値の設定に関する詳細な手順については、CREATE ROUTINE LOAD を参照してください。問題を特定してトラブルシューティングするために、ReasonOfStateChanged および ErrorLogUrls ファイルを確認できます。問題を修正した後、RESUME ROUTINE LOAD ステートメントを実行して PAUSED ロードジョブを再開できます。

ロードジョブの状態が CANCELLED の場合、ロードジョブが例外（たとえば、テーブルが削除された）に遭遇した可能性があります。問題を特定してトラブルシューティングするために、ReasonOfStateChanged および ErrorLogUrls ファイルを確認できます。ただし、CANCELLED ロードジョブを再開することはできません。

MySQL [example_db]> SHOW ROUTINE LOAD FOR example_tbl2_ordertest2 \G
*************************** 1. row ***************************
                  Id: 63013
                Name: example_tbl2_ordertest2
          CreateTime: 2022-08-10 17:09:00
           PauseTime: NULL
             EndTime: NULL
              DbName: default_cluster:example_db
           TableName: example_tbl2
               State: RUNNING
      DataSourceType: KAFKA
      CurrentTaskNum: 3
       JobProperties: {"partitions":"*","partial_update":"false","columnToColumnExpr":"commodity_id,customer_name,country,pay_time,pay_dt=from_unixtime(`pay_time`, '%Y%m%d'),price","maxBatchIntervalS":"20","whereExpr":"*","dataFormat":"json","timezone":"Asia/Shanghai","format":"json","json_root":"","strict_mode":"false","jsonpaths":"[\"$.commodity_id\",\"$.customer_name\",\"$.country\",\"$.pay_time\",\"$.price\"]","desireTaskConcurrentNum":"3","maxErrorNum":"0","strip_outer_array":"false","currentTaskConcurrentNum":"3","maxBatchRows":"200000"}
DataSourceProperties: {"topic":"ordertest2","currentKafkaPartitions":"0,1,2,3,4","brokerList":"<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>"}
    CustomProperties: {"kafka_default_offsets":"OFFSET_BEGINNING"}
           Statistic: {"receivedBytes":230,"errorRows":0,"committedTaskNum":1,"loadedRows":2,"loadRowsRate":0,"abortedTaskNum":0,"totalRows":2,"unselectedRows":0,"receivedBytesRate":0,"taskExecuteTimeMs":522}
            Progress: {"0":"1","1":"OFFSET_ZERO","2":"OFFSET_ZERO","3":"OFFSET_ZERO","4":"OFFSET_ZERO"}
ReasonOfStateChanged: 
        ErrorLogUrls: 
            OtherMsg: 

注意

停止したロードジョブやまだ開始されていないロードジョブを確認することはできません。

ロードタスクを確認する

SHOW ROUTINE LOAD TASK ステートメントを実行して、ロードジョブ example_tbl2_ordertest2 のロードタスクを確認します。現在実行中のタスクの数、消費される Kafka トピックパーティション、消費の進行状況 DataSourceProperties、および対応する Coordinator BE ノード BeId などを確認できます。

MySQL [example_db]> SHOW ROUTINE LOAD TASK WHERE JobName = "example_tbl2_ordertest2" \G
*************************** 1. row ***************************
              TaskId: 18c3a823-d73e-4a64-b9cb-b9eced026753
               TxnId: -1
           TxnStatus: UNKNOWN
               JobId: 63013
          CreateTime: 2022-08-10 17:09:05
   LastScheduledTime: 2022-08-10 17:47:27
    ExecuteStartTime: NULL
             Timeout: 60
                BeId: -1
DataSourceProperties: {"1":0,"4":0}
             Message: there is no new data in kafka, wait for 20 seconds to schedule again
*************************** 2. row ***************************
              TaskId: f76c97ac-26aa-4b41-8194-a8ba2063eb00
               TxnId: -1
           TxnStatus: UNKNOWN
               JobId: 63013
          CreateTime: 2022-08-10 17:09:05
   LastScheduledTime: 2022-08-10 17:47:26
    ExecuteStartTime: NULL
             Timeout: 60
                BeId: -1
DataSourceProperties: {"2":0}
             Message: there is no new data in kafka, wait for 20 seconds to schedule again
*************************** 3. row ***************************
              TaskId: 1a327a34-99f4-4f8d-8014-3cd38db99ec6
               TxnId: -1
           TxnStatus: UNKNOWN
               JobId: 63013
          CreateTime: 2022-08-10 17:09:26
   LastScheduledTime: 2022-08-10 17:47:27
    ExecuteStartTime: NULL
             Timeout: 60
                BeId: -1
DataSourceProperties: {"0":2,"3":0}
             Message: there is no new data in kafka, wait for 20 seconds to schedule again

ロードジョブを一時停止する

PAUSE ROUTINE LOAD ステートメントを実行して、ロードジョブを一時停止できます。ステートメントが実行されると、ロードジョブの状態は PAUSED になりますが、停止していません。RESUME ROUTINE LOAD ステートメントを実行して再開できます。SHOW ROUTINE LOAD ステートメントを使用して、そのステータスを確認することもできます。

次の例では、ロードジョブ example_tbl2_ordertest2 を一時停止します:

PAUSE ROUTINE LOAD FOR example_tbl2_ordertest2;

ロードジョブを再開する

RESUME ROUTINE LOAD ステートメントを実行して、一時停止されたロードジョブを再開できます。ロードジョブの状態は一時的に NEED_SCHEDULE（ロードジョブが再スケジュールされているため）になり、その後 RUNNING になります。SHOW ROUTINE LOAD ステートメントを使用して、そのステータスを確認できます。

次の例では、一時停止されたロードジョブ example_tbl2_ordertest2 を再開します:

RESUME ROUTINE LOAD FOR example_tbl2_ordertest2;

ロードジョブを変更する

ロードジョブを変更する前に、PAUSE ROUTINE LOAD ステートメントを使用して一時停止する必要があります。その後、ALTER ROUTINE LOAD を実行できます。変更後、RESUME ROUTINE LOAD ステートメントを実行して再開し、SHOW ROUTINE LOAD ステートメントを使用してそのステータスを確認できます。

生存している BE ノードの数が 6 に増加し、消費される Kafka トピックパーティションが "0,1,2,3,4,5,6,7" になったとします。実際のロードタスクの同時実行性を高めたい場合、次のステートメントを実行して、希望するタスクの同時実行数 desired_concurrent_number を 6 に増やし（生存している BE ノードの数以上）、Kafka トピックパーティションと初期オフセットを指定できます。

注意

実際のタスクの同時実行性は複数のパラメータの最小値によって決定されるため、FE の動的パラメータ max_routine_load_task_concurrent_num の値が 6 以上であることを確認する必要があります。

ALTER ROUTINE LOAD FOR example_tbl2_ordertest2
PROPERTIES
(
    "desired_concurrent_number" = "6"
)
FROM kafka
(
    "kafka_partitions" = "0,1,2,3,4,5,6,7",
    "kafka_offsets" = "OFFSET_BEGINNING,OFFSET_BEGINNING,OFFSET_BEGINNING,OFFSET_BEGINNING,OFFSET_END,OFFSET_END,OFFSET_END,OFFSET_END"
);

ロードジョブを停止する

STOP ROUTINE LOAD ステートメントを実行して、ロードジョブを停止できます。ステートメントが実行されると、ロードジョブの状態は STOPPED になり、停止したロードジョブを再開することはできません。SHOW ROUTINE LOAD ステートメントを使用して、停止したロードジョブのステータスを確認することはできません。

次の例では、ロードジョブ example_tbl2_ordertest2 を停止します:

STOP ROUTINE LOAD FOR example_tbl2_ordertest2;