MaxCompute:,

データの準備

• ローカルクライアント (odpscmd) または MaxCompute SQL コマンドを実行できる別のツールを使用して MaxCompute に接続し、ターゲットテーブルを作成します。

  CREATE TABLE IF NOT EXISTS table_text(
    topic STRING,
    `partition` BIGINT,
    `offset` BIGINT,
    key STRING,
    value STRING
  ) PARTITIONED BY (pt STRING);

• Kafka データの作成

  $KAFKA_HOME/bin/ ディレクトリで、次のコマンドを実行して Kafka トピックを作成します。この例では、トピック名として topic_text を使用します。

  sh kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_text

  次のコマンドを実行して Kafka メッセージを作成します。

  sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic topic_text --property parse.key=true
  >123    abc
  >456    edf

(任意) Kafka-connector サービスを起動します。詳細については、「Kafka コネクタサービスの設定と起動」をご参照ください。

odps-sink-connector.json ファイルを作成して設定します。次に、odps-sink-connector.json ファイルを $KAFKA_HOME/config パスなどの任意の場所にアップロードします。

次のコードは odps-sink-connector.json ファイルの例です。odps-sink-connector.json ファイルの詳細については、「Kafka コネクタタスクの設定と起動」をご参照ください。

{
    "name": "odps-test-text",
    "config": {
      "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
      "tasks.max": "3",
      "topics": "topic_text",
      "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
      "project": "project_name",
      "schema":"default",
      "table": "table_text",
      "account_type": "ALIYUN",
      "access_id": "<yourAccessKeyId>",
      "access_key": "<yourAccessKeySecret>",
      "partition_window_type": "MINUTE",
      "mode":"VALUE",
      "format":"TEXT",
      "sink_pool_size":"150",
      "record_batch_size":"9000",
      "buffer_size_kb":"600000"
    }
  }

次のコマンドを実行して、Kafka コネクタのデータ移行タスクを開始します。

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors -d @$KAFKA_HOME/config/odps-sink-connector.json

結果の検証

ローカルクライアント (odpscmd) または MaxCompute SQL コマンドを実行できる別のツールを使用して MaxCompute に接続し、次のコマンドを実行してデータをクエリし、結果を検証します。

set odps.sql.allow.fullscan=true;
select * from table_text;

次の出力が返されます：

# odps-sink-connector.json 設定ファイルの mode パラメーターが VALUE に設定されているため、値の内容のみが保持されます。キーフィールドは NULL です。

+-------+------------+------------+-----+-------+----+
| topic | partition  | offset     | key | value | pt |
+-------+------------+------------+-----+-------+----+
| topic_text | 0      | 0          | NULL | abc   | 07-13-2023 21:13 |
| topic_text | 0      | 1          | NULL | edf   | 07-13-2023 21:13 |
+-------+------------+------------+-----+-------+----+

データの準備

• ローカルクライアント (odpscmd) または MaxCompute SQL コマンドを実行できる別のツールを使用して MaxCompute に接続し、ターゲットテーブルを作成します。

  CREATE TABLE IF NOT EXISTS table_csv(
    topic STRING,
    `partition` BIGINT,
    `offset` BIGINT,
    id BIGINT,
    name STRING,
    region STRING
  ) PARTITIONED BY (pt STRING);

• Kafka へのデータの書き込み

  $KAFKA_HOME/bin/ ディレクトリで、次のコマンドを実行して topic_csv という名前の Kafka トピックを作成します。

  sh kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_csv

  次のコマンドを実行して Kafka メッセージを作成します。

  sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic topic_csv --property parse.key=true
  >123	1103,zhangsan,china
  >456	1104,lisi,usa

(任意) Kafka-connector サービスを起動します。詳細については、「Kafka コネクタサービスの設定と起動」をご参照ください。

odps-sink-connector.json ファイルを作成して設定し、odps-sink-connector.json ファイルを任意の場所にアップロードします。このトピックでは、例として $KAFKA_HOME/config パスを使用します。

次のコードは odps-sink-connector.json ファイルの例です。odps-sink-connector.json ファイルの詳細については、「Kafka コネクタタスクの設定と起動」をご参照ください。

{
    "name": "odps-test-csv",
    "config": {
      "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
      "tasks.max": "3",
      "topics": "topic_csv",
      "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
      "project": "project_name",    
      "schema":"default",
      "table": "table_csv",
      "account_type": "ALIYUN",
      "access_id": "<yourAccessKeyId>",
      "access_key": "<yourAccessKeySecret>",
      "partition_window_type": "MINUTE",
      "format":"CSV",
      "mode":"VALUE",
      "sink_pool_size":"150",
      "record_batch_size":"9000",
      "buffer_size_kb":"600000"
    }
  }

次のコマンドを実行して、Kafka コネクタのデータ移行タスクを開始します。

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors -d @$KAFKA_HOME/config/odps-sink-connector.json

結果の検証

ローカルクライアント (odpscmd) または MaxCompute SQL コマンドを実行できる別のツールを使用して MaxCompute に接続し、次のコマンドを実行してデータをクエリし、結果を検証します。

set odps.sql.allow.fullscan=true;
select * from table_csv;

次の出力が返されます：

+-------+------------+------------+------------+------+--------+----+
| topic | partition  | offset     | id         | name | region | pt |
+-------+------------+------------+------------+------+--------+----+
| csv_test | 0       | 0          | 1103       | zhangsan | china  | 07-14-2023 00:10 |
| csv_test | 0       | 1          | 1104       | lisi | usa    | 07-14-2023 00:10 |
+-------+------------+------------+------------+------+--------+----+

データの準備

• ローカルクライアント (odpscmd) または MaxCompute SQL コマンドを実行できる別のツールを使用して MaxCompute に接続し、ターゲットテーブルを作成します。

  CREATE TABLE IF NOT EXISTS table_json(
    topic STRING,
    `partition` BIGINT,
    `offset` BIGINT,
    key STRING,
    value JSON
  ) PARTITIONED BY (pt STRING);

• Kafka データの作成

  $KAFKA_HOME/bin/ ディレクトリで、次のコマンドを実行して Kafka トピックを作成します。この例では、トピック名として topic_json を使用します。

  sh kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_json

  次のコマンドを実行して、Kafka メッセージを作成します。

  sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic topic_json --property parse.key=true
  >123    {"id":123,"name":"json-1","region":"beijing"}                         
  >456    {"id":456,"name":"json-2","region":"hangzhou"}

(任意) Kafka-connector サービスを起動します。詳細については、「Kafka コネクタサービスの設定と起動」をご参照ください。

odps-sink-connector.json ファイルを作成して設定します。次に、odps-sink-connector.json ファイルを $KAFKA_HOME/config パスなどの任意の場所にアップロードします。

次のコードは odps-sink-connector.json ファイルの例です。odps-sink-connector.json ファイルの詳細については、「Kafka コネクタタスクの設定と起動」をご参照ください。

{
    "name": "odps-test-json",
    "config": {
      "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
      "tasks.max": "3",
      "topics": "topic_json",
      "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
      "project": "project_name",    
      "schema":"default",
      "table": "table_json",
      "account_type": "ALIYUN",
      "access_id": "<yourAccessKeyId>",
      "access_key": "<yourAccessKeySecret>",
      "partition_window_type": "MINUTE",
      "mode":"VALUE",
      "format":"JSON",
      "sink_pool_size":"150",
      "record_batch_size":"9000",
      "buffer_size_kb":"600000"
    }
  }

次のコマンドを実行して、Kafka コネクタのデータ移行タスクを開始します。

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors -d @$KAFKA_HOME/config/odps-sink-connector.json

結果の検証

ローカルクライアント (odpscmd) または MaxCompute SQL コマンドを実行できる別のツールを使用して MaxCompute に接続し、次のコマンドを実行してデータをクエリし、結果を検証します。

set odps.sql.allow.fullscan=true;
select * from table_json;

次の出力が返されます：

# JSON データは正常に value フィールドに書き込まれます。
+-------+------------+------------+-----+-------+----+
| topic | partition  | offset     | key | value | pt |
+-------+------------+------------+-----+-------+----+
| Topic_json | 0      | 0          | NULL | {"id":123,"name":"json-1","region":"beijing"} | 07-14-2023 00:28 |
| Topic_json | 0      | 1          | NULL | {"id":456,"name":"json-2","region":"hangzhou"} | 07-14-2023 00:28 |
+-------+------------+------------+-----+-------+----+

データの準備

• ローカルクライアント (odpscmd) または MaxCompute SQL コマンドを実行できる別のツールを使用して MaxCompute に接続し、ターゲットテーブルを作成します。

  CREATE TABLE IF NOT EXISTS table_flatten(
    topic STRING,
    `partition` BIGINT,
    `offset` BIGINT,
    id BIGINT,
    name STRING,
    extendinfo JSON
  ) PARTITIONED BY (pt STRING);

• Kafka データの作成

  $KAFKA_HOME/bin/ ディレクトリで、次のコマンドを実行して Kafka トピックを作成します。この例では、トピック名として topic_flatten を使用します。

  ./kafka/bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_flatten

  次のコマンドを実行して Kafka メッセージを作成します。

  sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic topic_flatten --property parse.key=true
  >123  {"id":123,"name":"json-1","extendinfo":{"region":"beijing","sex":"M"}}                         
  >456  {"id":456,"name":"json-2","extendinfo":{"region":"hangzhou","sex":"W"}}

(任意) Kafka-connector サービスを起動します。詳細については、「Kafka コネクタサービスの設定と起動」をご参照ください。

odps-sink-connector.json ファイルを作成して設定し、odps-sink-connector.json ファイルを任意の場所にアップロードします。このトピックでは、例として $KAFKA_HOME/config パスを使用します。

次のコードは odps-sink-connector.json ファイルの例です。odps-sink-connector.json ファイルの詳細については、「Kafka コネクタタスクの設定と起動」をご参照ください。

{
    "name": "odps-test-flatten",
    "config": {
      "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
      "tasks.max": "3",
      "topics": "topic_flatten",
      "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
      "project": "project_name",    
      "schema":"default",
      "table": "table_flatten",
      "account_type": "ALIYUN",
      "access_id": "<yourAccessKeyId>",
      "access_key": "<yourAccessKeySecret>",
      "partition_window_type": "MINUTE",
      "mode":"VALUE",
      "format":"FLATTEN",
      "sink_pool_size":"150",
      "record_batch_size":"9000",
      "buffer_size_kb":"600000"
    }
  }

次のコマンドを実行して、Kafka コネクタタスクを開始します。

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors -d @$KAFKA_HOME/config/odps-sink-connector.json

結果の検証

ローカルクライアント (odpscmd) または MaxCompute SQL コマンドを実行できる別のツールを使用して MaxCompute に接続し、次のコマンドを実行してデータをクエリし、結果を検証します。

set odps.sql.allow.fullscan=true;
select * from table_flatten;

次の結果が表示されます：

# JSON データは解析されて MaxCompute テーブルに書き込まれ、extendinfo はネストをサポートする JSON フィールドになります。
+-------+------------+--------+-----+------+------------+----+
| topic | partition  | offset | id  | name | extendinfo | pt |
+-------+------------+--------+-----+------+------------+----+
| topic_flatten | 0   | 0      | 123 | json-1 | {"sex":"M","region":"beijing"} | 07-14-2023 01:33 |
| topic_flatten | 0   | 1      | 456 | json-2 | {"sex":"W","region":"hangzhou"} | 07-14-2023 01:33 |
+-------+------------+--------+-----+------+------------+----+

データの準備

• ローカルクライアント (odpscmd) または MaxCompute SQL コマンドを実行できる別のツールを使用して MaxCompute に接続し、ターゲットテーブルを作成します。

  CREATE TABLE IF NOT EXISTS table_flatten(
    topic STRING,
    `partition` BIGINT,
    `offset` BIGINT,
    id BIGINT,
    name STRING,
    extendinfo JSON
  ) PARTITIONED BY (pt STRING);

• Kafka データの作成

  $KAFKA_HOME/bin/ ディレクトリで、次のコマンドを実行して Kafka トピックを作成します。

  • topic_abnormal トピック。

    sh kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_abnormal

  • runtime_error 例外のメッセージトピック。

    sh kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic runtime_error

    説明

    データ書き込み操作中にエラーが発生した場合、異常データは runtime_error トピックに書き込まれます。この種のエラーは通常、Kafka データと MaxCompute テーブルスキーマの不一致が原因で発生します。

  次のコマンドを実行して Kafka メッセージを作成します。

  次のコマンドのメッセージの1つが、ターゲットの MaxCompute テーブルのスキーマと一致しません。

  sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic flatten_test --property parse.key=true
  
  >100  {"id":100,"name":"json-3","extendinfo":{"region":"beijing","gender":"M"}}                         
  >101  {"id":101,"name":"json-4","extendinfos":"null"}
  >102	{"id":102,"name":"json-5","extendinfo":{"region":"beijing","gender":"M"}} 

(任意) Kafka-connector サービスを起動します。詳細については、「Kafka コネクタサービスの設定と起動」をご参照ください。

odps-sink-connector.json ファイルを作成して設定し、odps-sink-connector.json ファイルを任意の場所にアップロードします。このトピックでは、例として $KAFKA_HOME/config パスを使用します。

次のコードは odps-sink-connector.json ファイルの例です。odps-sink-connector.json ファイルの詳細については、「Kafka コネクタタスクの設定と起動」をご参照ください。

{
  "name": "odps-test-runtime-error",
  "config": {
    "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
    "tasks.max": "3",
    "topics": "topic_abnormal",
    "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
    "project": "project_name",
    "schema":"default",
    "table": "test_flatten",
    "account_type": "ALIYUN",
    "access_id": "<yourAccessKeyId>",
    "access_key": "<yourAccessKeySecret>",
    "partition_window_type": "MINUTE",
    "mode":"VALUE",
    "format":"FLATTEN",
    "sink_pool_size":"150",
    "record_batch_size":"9000",
    "buffer_size_kb":"600000",
    "runtime.error.topic.name":"runtime_error",
    "runtime.error.topic.bootstrap.servers":"http://XXXX",
    "skip_error":"false"
  }
}

次のコマンドを実行して、Kafka コネクタタスクを開始します。

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors -d @$KAFKA_HOME/config/odps-sink-connector.json

結果の検証

• MaxCompute テーブルデータのクエリ

  ローカルクライアント (odpscmd) または MaxCompute SQL コマンドを実行できる別のツールを使用して MaxCompute に接続し、次のコマンドを実行してデータをクエリし、結果を検証します。

  set odps.sql.allow.fullscan=true;
  select * from table_flatten;

  次の出力が返されます：

  # 結果からわかるように、ID 101 のデータはテーブルスキーマと一致しなかったため、MaxCompute に書き込まれませんでした。
  # runtime.error.topic.name パラメーターが設定されていたため、プロセスはブロックされず、後続のデータは正常に書き込まれました。
  +-------+------------+------------+------------+------+------------+----+
  | topic | partition  | offset     | id         | name | extendinfo | pt |
  +-------+------------+------------+------------+------+------------+----+
  | flatten_test | 0          | 0          | 123        | json-1 | {"gender":"M","region":"beijing"} | 07-14-2023 01:33 |
  | flatten_test | 0          | 1          | 456        | json-2 | {"gender":"W","region":"hangzhou"} | 07-14-2023 01:33 |
  | flatten_test | 0          | 0          | 123        | json-1 | {"gender":"M","region":"beijing"} | 07-14-2023 13:16 |
  | flatten_test | 0          | 1          | 456        | json-2 | {"gender":"W","region":"hangzhou"} | 07-14-2023 13:16 |
  | flatten_test | 0          | 2          | 100        | json-3 | {"gender":"M","region":"beijing"} | 07-14-2023 13:16 |
  | flatten_test | 0          | 4          | 102        | json-5 | {"gender":"M","region":"beijing"} | 07-14-2023 13:16 |
  +-------+------------+------------+------------+------+------------+----+

• runtime_error トピックのメッセージのクエリ

  $KAFKA_HOME/bin/ ディレクトリで、次のコマンドを実行してメッセージを表示します。

  sh kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic runtime_error --from-beginning

  次の結果が返されます：

  # 異常データは正常に runtime_error メッセージキューに書き込まれます。
  {"id":101,"name":"json-4","extendinfos":"null"}