E-MapReduce:Routine Load

前提条件

• 認証なし、または SSL 認証を使用する Kafka クラスターに接続できます。

• 次のメッセージ形式がサポートされています。

  • CSV 形式。各メッセージは、末尾に改行文字のない 1 行です。

  • JSON 形式。

• Array データ型はサポートされていません。

• Kafka 0.10.0.0 以降のみがサポートされています。

インポートジョブの作成

• 構文

  CREATE ROUTINE LOAD <database>.<job_name> ON <table_name>
      [COLUMNS TERMINATED BY "column_separator" ,]
      [COLUMNS (col1, col2, ...) ,]
      [WHERE where_condition ,]
      [PARTITION (part1, part2, ...)]
      [PROPERTIES ("key" = "value", ...)]
      FROM [DATA_SOURCE]
      [(data_source_properties1 = 'value1',
      data_source_properties2 = 'value2',
      ...)]

  次の表に、パラメーターの説明を示します。

  パラメーター	必須	説明
  job_name	はい	インポートジョブの名前。インポートデータベースの名前を先頭に付けることができます。名前は通常、「タイムスタンプ + テーブル名」の形式です。ジョブの名前は、データベース内で一意である必要があります。
  table_name	はい	インポート先テーブルの名前。
  COLUMNS TERMINATED 句	いいえ	ソースデータファイルの列区切り文字。デフォルト値：\t。
  COLUMNS 句	いいえ	ソースデータファイルの列とインポート先テーブルの列のマッピング。 マップされた列：ソースデータファイルの列とインポート先テーブルの列のマッピングを指定します。例えば、ソースデータの 1 番目の列をインポート先テーブルの col2 列に、2 番目の列を col1 列に、4 番目の列を col3 列にマッピングし、ソースデータの 3 番目の列をスキップしたい場合、句を COLUMNS (col2, col1, temp, col3) と記述します。ここで temp は、ソースデータの列をスキップするための一時的なプレースホルダーとして機能します。 派生列：StarRocks は、ソースデータファイルの列のデータを読み取るだけでなく、データ列に対する処理も可能です。たとえば、インポート先テーブルに列 col4 が追加され、col4 の値は、col1 の値と col2 の値を加算したものです。この場合、句は COLUMNS (col2, col1, temp, col3, col4 = col1 + col2) と記述できます。
  WHERE 句	いいえ	不要な行をフィルタリングするためのフィルター条件です。フィルター条件は、マップされた列または派生列で指定できます。 たとえば、k1 が 100 より大きく、k2 が 1000 に等しい行のみをインポートする場合、句は WHERE k1 > 100 and k2 = 1000 と記述できます。
  PARTITION 句	いいえ	インポート先テーブルのパーティション。パーティションを指定しない場合、ソースデータは自動的に対応するパーティションにインポートされます。
  PROPERTIES 句	いいえ	インポートジョブの共通パラメーター。
  desired_concurrent_number	いいえ	インポートジョブを分割できるタスクの最大数。値は 0 より大きい必要があります。デフォルト値：3。
  max_batch_interval	いいえ	各タスクの最大実行時間。有効値：5～60。単位：秒。デフォルト値：10。 V1.15 以降では、このパラメーターはタスクのスケジューリング時間を指定します。タスクの実行頻度を指定できます。fe.conf の routine_load_task_consume_second は、タスクのデータ消費時間を指定します。デフォルト値：3 秒。fe.conf の routine_load_task_timeout_second は、タスクの実行タイムアウト期間を指定します。デフォルト値：15 秒。
  max_batch_rows	いいえ	各タスクが読み取ることができる行の最大数。値は 200000 以上である必要があります。デフォルト値：200000。 V1.15 以降では、このパラメーターはエラー検出ウィンドウの範囲の定義にのみ使用されます。ウィンドウの範囲は 10 × max_batch_rows です。
  max_batch_size	いいえ	各タスクが読み取ることができる最大バイト数。単位：バイト。有効値：100 MB～1 GB。デフォルト値：100 MB。 V1.15 以降では、このパラメーターは廃止されました。fe.conf の routine_load_task_consume_second は、タスクのデータ消費時間を指定します。デフォルト値：3 秒。
  max_error_number	いいえ	サンプリングウィンドウで許可されるエラー行の最大数。値は 0 以上である必要があります。デフォルト値：0。エラー行は許可されません。 重要 WHERE 条件によってフィルターされた行はエラー行ではありません。
  strict_mode	いいえ	厳格モードを有効にするかどうかを指定します。デフォルトでは、このモードは有効になっています。 厳格モードを有効にすると、空でない生データが型変換の結果 NULL になった場合、そのデータはフィルタリングされます。厳格モードを無効にするには、このパラメーターを false に設定します。
  timezone	いいえ	インポートジョブのタイムゾーンです。 デフォルトでは、セッションの timezone パラメーターの値が使用されます。このパラメーターは、インポートに関連するすべてのタイムゾーン関連関数の結果に影響します。
  DATA_SOURCE	はい	データソースのタイプ。値を KAFKA に設定します。
  data_source_properties	いいえ	データソースに関する情報。値には次のフィールドが含まれます。 kafka_broker_list: Kafka ブローカーの接続情報です。形式は ip:host で、複数のブローカーはコンマ (,) で区切ります。 kafka_topic: サブスクライブする Kafka トピック。 説明 kafka_broker_list と kafka_topic フィールドは必須です。 kafka_partitions と kafka_offsets: サブスクライブする Kafka パーティションと各パーティションの開始オフセットです。 プロパティ: Kafka 関連のプロパティ。 このフィールドは、Kafka シェルの --property パラメーターに相当します。 HELP ROUTINE LOAD; コマンドを実行して、インポートジョブを作成するためのより詳細な構文を表示できます。

  説明

  HELP ROUTINE LOAD; コマンドを実行すると、インポートジョブを作成するためのより詳細な構文を表示できます。

• この例では、example_tbl2_ordertest という名前の Routine Load インポートジョブを作成します。このジョブは認証を使用せず、Kafka クラスターの ordertest2 トピックからメッセージを継続的に消費し、example_tbl テーブルにインポートします。ジョブは、指定されたパーティションで利用可能な最も早いオフセットから消費を開始します。

  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"
  );

• 例：次の例は、SSL を使用して Kafka への接続を構成する方法を示しています。

  -- セキュリティプロトコルとして SSL を指定します。
  "property.security.protocol" = "ssl", 
  
   -- CA 証明書の場所を指定します。
  "property.ssl.ca.location" = "FILE:ca-cert",
  
  -- Kafka サーバーでクライアント認証が有効になっている場合は、次の 3 つのパラメータも設定する必要があります。
  -- クライアントの公開鍵の場所。
  "property.ssl.certificate.location" = "FILE:client.pem", 
  -- クライアントの秘密鍵の場所。
  "property.ssl.key.location" = "FILE:client.key", 
  -- クライアントの秘密鍵のパスワード。
  "property.ssl.key.password" = "******"

  ファイルの作成の詳細については、CREATE FILE をご参照ください。

  説明

  CREATE FILE を使用する場合、url として Object Storage Service (OSS) の HTTP エンドポイントを使用します。詳細については、「IPv6 をサポートするエンドポイントを使用して OSS にアクセスする」をご参照ください。

ジョブステータスの表示

• load_test データベース内の、停止またはキャンセルされたジョブを含むすべての Routine Load ジョブを表示します。結果には 1 行以上が含まれる場合があります。

  USE load_test;
  SHOW ALL ROUTINE LOAD;

• load_test データベースで現在実行中の example_tbl2_ordertest という Routine Load ジョブを表示します。

  SHOW ROUTINE LOAD FOR load_test.example_tbl2_ordertest;

• E-MapReduce (EMR) StarRocks Manager コンソールで、メタデータ管理 をクリックします。ターゲットデータベースの名前をクリックし、Tasks をクリックして、[Kafka Import] タブでジョブステータスを表示します。

*************************** 1. row ***************************

                  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 row in set (0.00 sec)

インポートジョブの一時停止

PAUSE ステートメントを使用して、インポートジョブを一時停止します。ジョブは PAUSED 状態になり、データのインポートが中断されます。一時停止されたジョブは終了されず、RESUME ステートメントを使用して再開できます。

PAUSE ROUTINE LOAD FOR <job_name>;

インポートジョブを一時停止すると、State が PAUSED に変わり、Statistic と Progress の情報が更新されなくなります。ジョブは終了していないため、SHOW ROUTINE LOAD ステートメントを実行して表示できます。

インポートジョブの再開

一時停止中のジョブを再開するには、RESUME ステートメントを使用します。ジョブは一時的に NEED_SCHEDULE 状態になります。これは、ジョブが再スケジュール中であることを示します。その後、RUNNING 状態に戻り、データのインポートを続行します。

RESUME ROUTINE LOAD FOR <job_name>;

インポートジョブの停止

STOP ステートメントを使用して、インポートジョブを停止します。 ジョブは STOPPED 状態になり、データインポートが停止し、ジョブは終了します。 停止したジョブを再開することはできません。

STOP ROUTINE LOAD FOR <job_name>;