MaxCompute:オフラインモードまたはリアルタイムモードでMaxComputeにKafkaデータをインポートする

データを準備します。

• 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という名前のKafkaトピックが作成されます。

  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-connectorサービスの設定と開始」をご参照ください。

odps-sink-connector.jsonファイルを作成して構成し、odps-sink-connector.jsonファイルを任意の場所にアップロードします。 この例では、odps-sink-connector.jsonファイルは $KAFKA_HOME/configディレクトリにアップロードされます。

次のコードは、odps-sink-connector.jsonファイルの内容を示しています。 odps-sink-connector.jsonファイルの詳細については、「Kafka-connectorタスクの設定と開始」をご参照ください。

{
    "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-connectorタスクを開始します。

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を実行できる別のツールで次のコマンドを実行して、データ書き込み結果を照会します。

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

次の応答が返されます。

# The mode value in the odps-sink-connector.json configuration file is VALUE. Therefore, only the value is retained and the key field is 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 |
+-------+------------+------------+-----+-------+----+

データを準備します。

• MaxComputeクライアント (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/ ディレクトリで、次のコマンドを実行してKafkaトピックを作成します。 この例では、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-connectorサービスの設定と開始」をご参照ください。

odps-sink-connector.jsonファイルを作成して構成し、odps-sink-connector.jsonファイルを任意の場所にアップロードします。 この例では、odps-sink-connector.jsonファイルは $KAFKA_HOME/configディレクトリにアップロードされます。

次のコードは、odps-sink-connector.jsonファイルの内容を示しています。 odps-sink-connector.jsonファイルの詳細については、「Kafka-connectorタスクの設定と開始」をご参照ください。

{
    "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-connectorタスクを開始します。

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を実行できる別のツールで次のコマンドを実行して、データ書き込み結果を照会します。

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 |
+-------+------------+------------+------------+------+--------+----+

データを準備します。

• MaxComputeクライアント (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-connectorサービスの設定と開始」をご参照ください。

odps-sink-connector.jsonファイルを作成して構成し、odps-sink-connector.jsonファイルを任意の場所にアップロードします。 この例では、odps-sink-connector.jsonファイルは $KAFKA_HOME/configディレクトリにアップロードされます。

次のコードは、odps-sink-connector.jsonファイルの内容を示しています。 odps-sink-connector.jsonファイルの詳細については、「Kafka-connectorタスクの設定と開始」をご参照ください。

{
    "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-connectorタスクを開始します。

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を実行できる別のツールで次のコマンドを実行して、データ書き込み結果を照会します。

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

次の応答が返されます。

# Write JSON data to the value field.
+-------+------------+------------+-----+-------+----+
| 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 |
+-------+------------+------------+-----+-------+----+

データを準備します。

• MaxComputeクライアント (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-connectorサービスの設定と開始」をご参照ください。

odps-sink-connector.jsonファイルを作成して構成し、odps-sink-connector.jsonファイルを任意の場所にアップロードします。 この例では、odps-sink-connector.jsonファイルは $KAFKA_HOME/configディレクトリにアップロードされます。

次のコードは、odps-sink-connector.jsonファイルの内容を示しています。 odps-sink-connector.jsonファイルの詳細については、「Kafka-connectorタスクの設定と開始」をご参照ください。

{
    "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-connectorタスクを開始します。

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を実行できる別のツールで次のコマンドを実行して、データ書き込み結果を照会します。

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

次の応答が返されます。

# JSON data is parsed and written to the MaxCompute table. The exteninfo field in the nested JSON format can be a JSON field.
+-------+------------+--------+-----+------+------------+----+
| 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 |
+-------+------------+--------+-----+------+------------+----+

データを準備します。

• MaxComputeクライアント (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トピックを作成するコマンドを実行します。

  • トピック_異常

    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-connectorサービスの設定と開始」をご参照ください。

odps-sink-connector.jsonファイルを作成して構成し、odps-sink-connector.jsonファイルを任意の場所にアップロードします。 この例では、odps-sink-connector.jsonファイルは $KAFKA_HOME/configディレクトリにアップロードされます。

次のコードは、odps-sink-connector.jsonファイルの内容を示しています。 odps-sink-connector.jsonファイルの詳細については、「Kafka-connectorタスクの設定と開始」をご参照ください。

{
  "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-connectorタスクを開始します。

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テーブルのデータを照会します。

  MaxComputeクライアント (odpscmd) またはMaxCompute SQLを実行できる別のツールで次のコマンドを実行して、データ書き込み結果を照会します。

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

  次の応答が返されます。

  # The last two records are displayed. This is because the skip_error parameter is set to true. The data with the id of 101 is not written to the MaxCompute table, and subsequent records are not blocked from being written to the MaxCompute table. 
  +-------+------------+------------+------------+------+------------+----+
  | 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

  次の応答が返されます。

  # Abnormal data is written to the runtime_error topic.
  {"id":101,"name":"json-4","extendinfos":"null"}