狠狠躁天天躁中文字幕,最新熟女中文字幕97,好看日韩无码AV黄色网站

MaxCompute與Kafka的集成能夠提供高效、可靠的數據處理和分析能力，適用于需要實時處理、大規模數據流和復雜數據分析的場景。本文介紹消息隊列Kafka版和自建Kafka數據的寫入流程，以及自建Kafka數據的寫入示例。

Kafka數據寫入MaxCompute流程：阿里云全托管Kafka

MaxCompute與消息隊列Kafka版服務緊密集成，借助消息隊列Kafka版服務的MaxCompute Sink Connector，無需第三方工具及二次開發，即可滿足將指定Topic數據持續導入MaxCompute數據表的需求，操作詳情請參見創建MaxCompute Sink Connector。

Kafka數據寫入MaxCompute流程：自建開源Kafka

前提條件

已部署V2.2及以上版本的Kafka服務（推薦最新版本V3.4.0），并已創建Kafka Topic信息。
已創建MaxCompute項目和表。具體操作，請參見創建MaxCompute項目和創建表。

注意事項

Kafka-connector服務支持TEXT、CSV、JSON和FLATTEN類型的Kafka數據寫入，不同類型的注意事項詳情如下。關于數據類型的詳情介紹，請參見數據類型說明。

TEXT和JSON類型的Kafka數據寫入MaxCompute時，MaxCompute表要求如下：

字段名稱	字段類型	是否為固定字段
topic	STRING	是
partition	BIGINT	是
offset	BIGINT	是
key	TEXT類型Kafka數據寫入時，字段類型必須為STRING。 JSON類型Kafka數據寫入時，根據寫入的數據類型設置，支持STRING與JSON。	需要將Kafka消息的中的Key值同步到MaxCompute表中時，此字段為固定字段。關于Kafka消息同步到MaxCompute的模式，詳情請參見mode。
value	TEXT類型Kafka數據寫入時，字段類型必須為STRING。 JSON類型Kafka數據寫入時，根據寫入的數據類型設置，支持STRING與JSON。	需要將Kafka消息的中的Value值同步到MaxCompute表中時，此字段為固定字段。關于Kafka消息同步到MaxCompute的模式，詳情請參見mode。
pt	STRING（分區字段）	是

FLATTEN和CSV類型的Kafka數據寫入MaxCompute時，必須包含以下字段和字段類型，您可以根據寫入數據的內容自定義其他字段。
字段名稱
字段類型
topic
STRING
partition
BIGINT
offset
BIGINT
pt
STRING（分區字段）
- CSV類型的Kafka數據寫入MaxCompute表中時，MaxCompute表中自定義的字段順序和字段類型，必須與Kafka寫入的數據保持一致，以確保數據能正確寫入。
- FLATTEN類型的Kafka數據寫入MaxCompute表中時，MaxCompute表中自定義的字段名稱必須Kafka數據中字段名稱保持一致，以確保數據能正確寫入。
  例如：要寫入的FLATTEN類型的Kafka數據內容為{"A":a,"B":"b","C":{"D":"d","E":"e"}}，那MaxCompute表信息如下所示。
```
CREATE TABLE IF NOT EXISTS table_flatten(
 topic STRING,
 `partition` BIGINT,
 `offset` BIGINT,
 A BIGINT,
 B STRING,
 C JSON
) PARTITIONED BY (pt STRING);
```

配置并啟動Kafka-connector服務

以Linux環境為例，在命令窗口執行以下命令或下載鏈接，下載kafka-connector-2.0.jar包。
```
wget http://maxcompute-repo.oss-cn-hangzhou.aliyuncs.com/kafka/kafka-connector-2.0.jar
```
為防止依賴沖突，建議在$KAFKA_HOME/libs下新建一個子文件夾，例如connector，用來放置kafka-connector-2.0.jar包。
說明
若kafka-connector-2.0.jar包與Kafka的部署環境不一致，配置并啟動Kafka-connector服務的操作詳情，請參見配置Kafka-connector。

在$KAFKA_HOME/config目錄下，配置connect-distributed.properties文件。

在connect-distributed.properties文件中補充以下內容。

##新增以下內容
plugin.path=<KAFKA_HOME>/libs/connector

##更新key.converter和value.converter參數值
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter

在$KAFKA_HOME/路徑下，執行以下命令，啟動Kafka-connector服務。

##啟動命令
bin/connect-distributed.sh config/connect-distributed.properties &

配置并啟動Kafka-connector任務

創建并配置odps-sink-connector.json配置文件，并將odps-sink-connector.json文件上傳至任意位置。

odps-sink-connector.json配置文件內容與參數介紹如下。

{
  "name": "Kafka connector task name",
  "config": {
    "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
    "tasks.max": "3",
    "topics": "your_topic",
    "endpoint": "endpoint",
    "tunnel_endpoint": "your_tunnel endpoint",
    "project": "project",
    "schema":"default",
    "table": "your_table",
    "account_type": "account type (STS or ALIYUN)",
    "access_id": "access id",
    "access_key": "access key",
    "account_id": "account id for sts",
    "sts.endpoint": "sts endpoint",
    "region_id": "region id for sts",
    "role_name": "role name for sts",
    "client_timeout_ms": "STS Token valid period (ms)",
    "format": "TEXT",
    "mode": "KEY",
    "partition_window_type": "MINUTE",
    "use_streaming": false,
    "buffer_size_kb": 65536,
    "sink_pool_size":"150",
    "record_batch_size":"8000",
    "runtime.error.topic.name":"kafka topic when runtime errors happens",
    "runtime.error.topic.bootstrap.servers":"kafka bootstrap servers of error topic queue",
    "skip_error":"false"
  }
}

公共參數

參數名	是否必填	說明
name	是	任務名稱，且名稱必須保持唯一。
connector.class	是	啟動`Kafka connector`服務的類名，默認值為`com.aliyun.odps.kafka.connect.MaxComputeSinkConnector`。
tasks.max	是	`Kafka connector`中消費者進程最大個數，必須為大于0的整數。
topics	是	Kafka的Topic名稱。
endpoint	是	MaxCompute服務的連接地址。您需要根據創建MaxCompute項目時選擇的地域以及網絡連接方式配置Endpoint。各地域及網絡對應的Endpoint值，請參見Endpoint。
tunnel_endpoint	否	Tunnel服務的外網訪問鏈接。如果您未配置Tunnel Endpoint，Tunnel會自動路由到MaxCompute服務所在網絡對應的Tunnel Endpoint。如果您配置了Tunnel Endpoint，則以配置為準，不進行自動路由。各地域及網絡對應的Tunnel Endpoint值，請參見Endpoint。
project	是	訪問的目標MaxCompute項目名稱。
schema	否	若目標MaxCompute項目配置Schema三層模型，則需要此參數，且默認值為default。若目標MaxCompute項目未配置Schema三層模型，則無需配置此參數。關于Schema的介紹詳情，請參見Schema操作。
table	是	目標MaxCompute項目的表名稱。
format	否	寫入的消息格式。取值如下： TEXT（默認值）：消息的格式為字符串。 BINARY：消息的格式為字節數組。 CSV：消息的格式為逗號（,）分隔的字符串。 JSON：消息格式為JSON數據類型的字符串。關于MaxCompute JSON類型的詳情，請參見MaxCompute JSON類型使用指南（試用Beta版本）。 FLATTEN：消息格式為JSON數據類型的字符串，JSON中的Key和Value會被解析，寫入到對應的MaxCompute表中，其中JSON數據中的Key和需要與MaxCompute的表列名對應。關于不同格式消息導入的案例，詳情請參見使用示例。
mode	否	消息同步到MaxCompute的模式。取值說明如下： KEY：只保留消息的Key，并將Key值寫入目標MaxCompute表中。 VALUE：只保留消息的Value，并將Value值寫入目標MaxCompute表中。 DEFAULT（默認值）：同時保留消息的Key和Value，并將Key和Value值都寫入目標MaxCompute表中。 DEFAULT模式下，只支持TEXT和BINARY格式數據寫入。
partition_window_type	否	按照系統時間進行數據分區。取值為DAY、HOUR（默認值）、MINUTE。
use_streaming	否	是否使用流式數據通道。取值說明如下： false（默認值）：不使用。 true：使用。
buffer_size_kb	否	odps partition writer內部緩沖區的大小，單位KB。默認65536 KB。
sink_pool_size	否	多線程寫入的最大線程數，默認為系統CPU核數。
record_batch_size	否	一個Kafka-connector任務內部的一個線程最多可以一次并行發送消息數量。
skip_error	否	是否跳過發生未知錯誤的記錄。取值說明如下： false（默認值）：不會跳過。 true：跳過。說明當skip_error為false且未配置runtime.error.topic.name參數，若遇到未知錯誤，會停止后續的數據寫入，進程會被阻塞并在日志中拋出異常。當skip_error取值true且runtime.error.topic.name未配置，寫入數據的進程會繼續寫入，異常數據會被丟棄。當skip_error為false且已配置runtime.error.topic.name參數，寫入數據的進程會繼續寫入，異常數據會被記錄到runtime.error.topic.name配置的Topic中。異常數據處理示例詳情，請參見異常數據處理示例。
runtime.error.topic.name	否	將數據寫入時發生的未知錯誤的數據寫入至Kafka的Topic名稱。
runtime.error.topic.bootstrap.servers	否	將數據寫入時發生的未知錯誤的數據寫入至Kafka的bootstrap servers地址。
account_type	是	訪問目標MaxCompute服務的方式，支持STS、ALIYUN兩種方式，默認ALIYUN。不同方式訪問MaxCompute需要配置不同的訪問憑證參數，詳情請參見通過ALIYUN方式訪問MaxCompute和通過STS方式訪問MaxComput。

通過ALIYUN方式訪問MaxCompute，除公共參數外還需配置以下參數。

參數名

說明

access_id

阿里云賬號或RAM賬號的AccessKey ID。

您可以進入AccessKey管理頁面獲取AccessKey ID。

access_key

AccessKey ID對應的AccessKey Secret。

您可以進入AccessKey管理頁面獲取AccessKey Secret。

通過STS方式訪問MaxCompute，除公共參數外還需配置以下參數。

參數名	說明
account_id	訪問目標MaxCompute項目的賬號ID。您可以進入賬號中心查看您的賬號ID。
region_id	訪問目標MaxCompute項目的地域ID。各地域對應的地域ID，請參見服務接入點。
role_name	訪問目標MaxCompute項目的角色名稱。您可以進入角色頁面查看角色名稱。
client_timeout_ms	STS Token刷新的時間間隔，單位為毫秒（ms），默認值為11（ms）。
sts.endpoint	使用臨時安全令牌（STS）進行身份認證時需要的STS 服務地址。各地域及網絡對應的Endpoint值，請參見服務接入點。

執行以下命令，啟動Kafka-connector數據傳輸任務。

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

TEXT類型數據寫入

數據準備。

通過使用本地客戶端（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。以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-connector服務。
說明
若Kafka-connector服務已啟動，可跳過此步驟。

創建并配置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": "LTAI5tM2iHkTd4W69nof****",
      "access_key": "S0uZvwDYDa56WZ1tjVmA67z1YS****",
      "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

結果驗證。

通過使用本地客戶端（odpscmd）連接或其他可以運行MaxCompute SQL的工具，執行如下命令，查詢數據寫入結果。

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

返回結果如下：

# 這里由于我們odps-sink-connector.json配置文件中的mode值為VALUE，所以只保留value的內容，key字段為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 |
+-------+------------+------------+-----+-------+----+

CSV類型數據寫入

數據準備。

通過使用本地客戶端（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。以topic_csv為例。

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服務。
說明
若Kafka-connector服務已啟動，可跳過此步驟。

創建并配置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": "LTAI5tM2iHkTd4W69nof****",
      "access_key": "S0uZvwDYDa56WZ1tjVmA67z1YS****",
      "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

結果驗證。

通過使用本地客戶端（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 |
+-------+------------+------------+------------+------+--------+----+

JSON類型數據寫入

數據準備。

通過使用本地客戶端（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。以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服務。
說明
若Kafka-connector服務已啟動，可跳過此步驟。

創建并配置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": "LTAI5tM2iHkTd4W69nof****",
      "access_key": "S0uZvwDYDa56WZ1tjVmA67z1YS****",
      "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

結果驗證。

通過使用本地客戶端（odpscmd）連接或其他可以運行MaxCompute SQL的工具，執行如下命令，查詢數據寫入結果。

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

FLATTEN類型數據寫入

數據準備。

通過使用本地客戶端（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。以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服務。
說明
若Kafka-connector服務已啟動，可跳過此步驟。

創建并配置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": "LTAI5tM2iHkTd4W69nof****",
      "access_key": "S0uZvwDYDa56WZ1tjVmA67z1YS****",
      "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

結果驗證。

通過使用本地客戶端（odpscmd）連接或其他可以運行MaxCompute SQL的工具，執行如下命令，查詢數據寫入結果。

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

返回結果如下：

# json數據被解析寫入MaxCompute表中，且支持json嵌套類型exteninfo為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。

topic_abnormalTopic。

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

runtime_error異常消息Topic。
```
sh kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic runtime_error
```
說明
當數據寫入發生未知錯誤（通常是Kafka數據與MaxCompute表格式不匹配），異常數據會被寫入到runtime_errorTopic中。

執行以下命令，創建Kafka消息。

以下消息中，其中一條數據格式與目標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服務。
說明
若Kafka-connector服務已啟動，可跳過此步驟。

創建并配置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": "LTAI5tM2iHkTd4W69nof****",
    "access_key": "S0uZvwDYDa56WZ1tjVmA67z1YS****",
    "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表數據

通過使用本地客戶端（odpscmd）連接或其他可以運行MaxCompute SQL的工具，執行如下命令，查詢數據寫入結果。

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

返回結果如下：

# 我們看到最后兩條數據，因為設置了skip_error參數為true,所以id為101的數據沒有被寫入MaxCompute，且沒有block后面數據的寫入。
+-------+------------+------------+------------+------+------------+----+
| 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_errorTopic的消息

在$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"}

日本熟妇hd丰满老熟妇,中文字幕一区二区三区在线不卡 ,亚洲成片在线观看,免费女同在线一区二区

使用Kafka（離線與實時）

Kafka數據寫入MaxCompute流程：阿里云全托管Kafka

Kafka數據寫入MaxCompute流程：自建開源Kafka

前提條件

注意事項

配置并啟動Kafka-connector服務

配置并啟動Kafka-connector任務

使用示例

TEXT類型數據寫入

CSV類型數據寫入

JSON類型數據寫入

FLATTEN類型數據寫入

異常數據處理示例