即時消費Kafka資料

Apache Kafka是一個容錯、低延遲、分布式的發布-訂閱訊息系統。Streaming Server支援從Apache和Confluent Kafka發行版中載入Kafka資料。通過雲原生資料倉儲AnalyticDB PostgreSQL版可讀外表對Kafka資料進行轉換，並將資料寫入雲原生資料倉儲AnalyticDB PostgreSQL版目標表中。

前提條件

• Kafka服務與雲原生資料倉儲AnalyticDB PostgreSQL版執行個體需在同一Virtual Private Cloud。

  重要

  如果Kafka服務與雲原生資料倉儲AnalyticDB PostgreSQL版執行個體屬於同一專用網路但是不在同一交換器（vSwitch）下，您需要進行如下操作：

  • 將Kafka服務所在交換器的IPv4網段添加至雲原生資料倉儲AnalyticDB PostgreSQL版執行個體白名單中。具體操作，請參見設定白名單。

  • 將雲原生資料倉儲AnalyticDB PostgreSQL版執行個體所在交換器的IPv4網段添加至Kafka服務白名單中。具體操作，請參見配置白名單。

• 已在Kafka服務中產生了大量範例資料。本文以阿里雲雲訊息佇列Kafka版為例，具體資訊如下。

  • 存取點資訊：alikafka-post-cn-wwo3hflb**-1-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-wwo3hflb-2-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-wwo3hflb**-3-vpc.alikafka.aliyuncs.com:9092

  • Topic：test_topic

  • consumer group：test_consumer_group

• 已在雲原生資料倉儲AnalyticDB PostgreSQL版中建立目標使用者和目標表，同時在任務中使用的資料庫使用者需要具備以下許可權。

  • 使用gpfdist協議建立唯讀外表的許可權。

  • 任務中配置的資料庫Schema的USAGE和CREATE許可權。

  • 任務中配置的寫入目標表的SELECT和INSERT許可權。

  本文以liss_test使用者和liss_test.liss_test_plaintext表為例。

  CREATE role liss_test with login;
  ALTER role liss_test with password 'lissTest****';
  ALTER role liss_test CREATEEXTTABLE(type = 'readable', protocol = 'gpfdist'); 
  \c - liss_test
  CREATE DATABASE liss_test;
  \c liss_test
  CREATE SCHEMA liss_test;
  CREATE TABLE liss_test.liss_test_plaintext (
  column_1 varchar(32),
  column_2 bigint,
  column_3 numeric,
  column_4 varchar(32),
  column_5 varchar(32)
  ) distributed by (column_1, column_2);

使用限制

• AnalyticDB for PostgreSQL6.0執行個體需為v6.6.0及以上版本。AnalyticDB for PostgreSQL7.0執行個體需為v7.0.3及以上版本。AnalyticDB PostgreSQL版Serverless模式執行個體暫不支援。

• 即時資料消費目前僅支援INSERT、MERGE（UPSERT）、UPDATE三種文法，暫不支援DELETE與READ。

• 使用MERGE（UPSERT）或UPDATE時，需要AnalyticDB for PostgreSQL表有主鍵索引。

• 使用即時資料消費，不同分區（Partition）之間需要使用主鍵列做分區因子，否則可能會造成全域死結錯誤，導致部分資料更新失敗。

• 即時資料消費當前僅支援Kafka訊息佇列，暫不支援CDC格式的資料來源。

• 當前的版本嚮導模式支援CSV和Delimited兩種格式的資料來源，專業模式支援CSV、Delimited和protobuf三種格式的資料來源。

操作步驟

步驟一：開啟即時資料服務

1. 登入雲原生資料倉儲AnalyticDB PostgreSQL版控制台。

2. 在控制台左上方，選擇執行個體所在地區。

3. 找到目標執行個體，單擊執行個體ID。

4. 在控制台左側導覽列單擊实时数据消费，再單擊左上方开启实时数据服务。

5. 在彈出的對話方塊中填寫名称及服务描述並單擊确定。開通完成後，可在控制台看到服務状态和连接信息。

   說明

   服務規格當前不可選，預設為8CU。

步驟二：新增即時資料源

1. 登入雲原生資料倉儲AnalyticDB PostgreSQL版控制台。

2. 在控制台左上方，選擇執行個體所在地區。

3. 找到目標執行個體，單擊執行個體ID。

4. 在左側導覽列，單擊实时数据消费。

5. 在实时数据源卡片中，單擊新增数据源，並完成以下配置。

   配置項	描述
   關聯資料服務	在下拉框中選擇已建立的即時資料服務。
   資料來源名稱	自訂資料來源名稱。
   資料來源描述	自訂資料來源描述。
   資料來源類型	目前僅支援Kafka。
   brokers	Kafka存取點資訊。 阿里雲的Kafka服務，可登入阿里雲控制台擷取預設存取點。具體操作，請參見查看存取點。 自建的kafka服務，Brokers需要填寫Kafka服務具體的`hostname:port`或`ip:port`資訊。
   topic	Kafka的Topic名稱。
   format	目前的版本嚮導模式支援CSV和Delimited兩種格式的資料來源，專業模式支援CSV、Delimited和protobuf三種格式的資料來源。
   資料行分隔符號	可設定任意單字元分隔字元。

6. 單擊确定。

步驟三：新增即時任務

1. 在实时任务卡片中，單擊新增实时任务，並完成以下配置。

   請根據業務需要選擇向导模式或专业模式。

   嚮導模式：可以通過控制台中的指引來快速搭建任務。

   專業模式：可以通過提交YAML的方式向Streaming Server提交任務，功能相比於嚮導模式更豐富。

   嚮導模式

   配置項		描述
   基本資料	任務名稱	定義任務的名稱，任務名稱不可以重複，必填。
   	任務描述	描述任務內容，選填。
   	配置模式	嚮導模式。
   源端配置	資料來源	選擇在新增即時資料源中配置的資料來源，目前僅支援Kafka為源的資料來源。
   	group_name	Kafka的消費者組。
   	failback_offset	消費位點。 earliest：從最早可用位點消費。 latest：從最新的位點開始消費。
   	投遞保證	Realtime Compute中的一致性語義，支援： ATLEAST：在Kafka中的資料至少有一次被寫入雲原生資料倉儲AnalyticDB PostgreSQL版。 EXACTLY：在Kafka中的資料有且僅有一次被寫入雲原生資料倉儲AnalyticDB PostgreSQL版。
   目標端配置	目標庫	需要寫入的雲原生資料倉儲AnalyticDB PostgreSQL版目標資料庫名稱。
   	Schema	雲原生資料倉儲AnalyticDB PostgreSQL版的模式名稱。
   	目標表	需要寫入的雲原生資料倉儲AnalyticDB PostgreSQL版目標表名稱。
   	帳號	當前任務使用的雲原生資料倉儲AnalyticDB PostgreSQL版資料庫帳號。
   	密碼	帳號密碼。
   	寫入模式	目前僅支援INSERT、UPDATE和MERGE三種寫入模式。 INSERT：將資料直接寫入目標表。 UPDATE：當輸入列中的MatchColumns與目標表中的列匹配，更新UpdateColumns中列出的目標表列。 MERGE：當寫入資料與目標表列的值相等時，使用寫入資料更新目標表列的現有資料。當寫入資料與目標表列的值不相等時，直接將資料寫入目標表。MERGE寫入模式可類比於UPSERT（UPDATE and INSERT），關於UPSERT的寫入方式，請參見使用INSERT ON CONFLICT覆蓋寫入資料。 說明 MatchColumns與UpdateColumns的含義請參見下文欄位類型的描述。
   	ErrorLimitCount	錯誤資料的容忍閾值。當寫入的錯誤資料到達ErrorLimitCount時，Streaming Server會自動停止將資料來源的資料寫入雲原生資料倉儲AnalyticDB PostgreSQL版。0表示Streaming Server遇到第一次錯誤資料時就會停止資料寫入。目前該參數未啟用，填0即可。
   欄位對應	源欄位	Kafka訊息中的Value欄位名，需要按照在Value中出現的順序指定所有的欄位名。
   	目標欄位	雲原生資料倉儲AnalyticDB PostgreSQL版目標表的欄位名。
   	欄位類型	目前支援以下三種類型： MatchColumns：作為寫入時的Join條件列作為更新條件，用於判斷目標表中哪些行需要被更新。 UpdateColumns：如果某一行資料符合更新條件，那麼在UpdateColumns中的列會被更新為新的值。 空（不填）：即使某一行資料符合更新條件，該欄位也不會被更新為新的值。 在UPDATE和MERGE寫入時，Streaming Server會先將資料寫入一個暫存資料表，然後利用MatchColumns作為條件列與目標表進行Join： 如果有匹配的資料，則會更新UpdateColumns中的資料。 如果沒有匹配的資料時，則會根據寫入模式有以下兩種情況： 寫入模式為UPDATE時，資料不會被寫入。 寫入模式為MERGE時，資料會被寫入。

   專業模式

   配置項		描述
   基本資料	任務名稱	定義任務的名稱，任務名稱不可以重複，必填。
   	任務描述	描述任務內容，選填。
   	配置模式	專業模式。
   	資料來源	選擇在新增即時資料源中配置的資料來源，目前僅支援Kafka為源的資料來源。
   YAML		可以通過YAML配置更複雜的寫入邏輯。本文的YAML配置樣本如下。更多詳情，請參見附錄：YAML配置說明。 DATABASE: liss_test USER: liss_test PASSWORD: lissTest**** HOST: gp-2ze517f9l7****-master.gpdb.rds-aliyun-pre.rds.aliyuncs.com PORT: 5432 KAFKA: INPUT: SOURCE: BROKERS: alikafka-post-cn-wwo3hflbo002-1-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-wwo3hflbo002-2-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-wwo3hflbo002-3-vpc.alikafka.aliyuncs.com:9092 TOPIC: test_topic FALLBACK_OFFSET: EARLIEST VALUE: COLUMNS: - NAME: column_1 TYPE: varchar(32) - NAME: column_2 TYPE: bigint - NAME: column_3 TYPE: numeric - NAME: column_4 TYPE: varchar(32) - NAME: column_5 TYPE: varchar(32) FORMAT: delimited DELIMITED_OPTION: DELIMITER: "|" ERROR_LIMIT: 20 OUTPUT: SCHEMA: liss_test TABLE: liss_test_plaintext MODE: MERGE MATCH_COLUMNS: - column_1 - column_2 UPDATE_COLUMNS: - column_3 - column_4 - column_5 MAPPING: - NAME: column_1 EXPRESSION: column_1 - NAME: column_2 EXPRESSION: column_2 - NAME: column_3 EXPRESSION: column_3 - NAME: column_4 EXPRESSION: column_4 - NAME: column_5 EXPRESSION: column_5 COMMIT: MAX_ROW: 1000 MINIMAL_INTERVAL: 1000 CONSISTENCY: ATLEAST POLL: BATCHSIZE: 1000 TIMEOUT: 1000 PROPERTIES: group.id: test_consumer_group

2. 單擊确定，並等待即時任務狀態為运行中，即可將資料來源中的資料寫入雲原生資料倉儲AnalyticDB PostgreSQL版。

在任務啟動後會在目標端配置的Schema（專業模式為METADATA.SCHEMA中配置的schema）下產生任務的兩種輔助表，其格式分別為：

• lissext_$UID：本任務定義的gpfdist外表，用於將資料寫入雲原生資料倉儲AnalyticDB PostgreSQL版。

• lisskafka_mission_info_$UID：用於儲存任務當前位點推進的情況，保障資料寫入的一致性。目前為了保障寫入任務的高可用，每個寫入任務會產生4個子任務，所以每啟動一個寫入任務，會產生4張表。

• UID是每個寫入任務的唯一標識ID。

附錄：YAML配置說明

YAML設定檔格式如下。

DATABASE: <db_name>
USER: <user_name>
PASSWORD: <password>
HOST: <host>
PORT: <adbpg_port>
VERSION: 2
KAFKA:
   INPUT:
      SOURCE:
        BROKERS: <kafka_broker_host:broker_port> [, ... ]
        TOPIC: <kafka_topic>
        [PARTITIONS: (<partition_numbers>)]
        [FALLBACK_OFFSET: { earliest | latest }]
      [VALUE:
        COLUMNS:
           - NAME: { <column_name> }
             TYPE: <column_data_type>
           [ ... ]
         FORMAT: <value_data_format>
         [[DELIMITED_OPTION:
            DELIMITER: <delimiter_string>
            [QUOTE: <quote_char>]
            [ESCAPE: <escape_char>] ] |
         [CSV_OPTION:
            [DELIMITER: <delim_char>]
            [QUOTE: <quote_char>]
            [NULL_STRING: <nullstr_val>]
            [ESCAPE: <escape_char>]
      [KEY:
        COLUMNS:
           - NAME: { <column_name> }
             TYPE: <column_data_type>
           [ ... ]
         FORMAT: <key_data_format>
         [[DELIMITED_OPTION:
            DELIMITER: <delimiter_string> |
            [QUOTE: <quote_char>]
            [ESCAPE: <escape_char>] ] |
         [CSV_OPTION:
            [DELIMITER: <delim_char>]
            [QUOTE: <quote_char>]
            [NULL_STRING: <nullstr_val>]
            [ESCAPE: <escape_char>]
      [META:
         COLUMNS:
            - NAME: <meta_column_name>
              TYPE: { json | jsonb }
         FORMAT: json]
      [ERROR_LIMIT: { <num_errors> | <percentage_errors> }]
   { OUTPUT:
      [SCHEMA: <output_schema_name>]
      TABLE: <table_name>
      [MODE: <mode>]
      [MATCH_COLUMNS: 
         - <match_column_name>
         [ ... ]]
      [ORDER_COLUMNS: 
         - <order_column_name>
         [ ... ]]
      [UPDATE_COLUMNS: 
         - <update_column_name>
         [ ... ]]
      [MAPPING: 
         - NAME: <target_column_name>
           EXPRESSION: { <source_column_name> | <expression> } 
         [ ... ]
           |
         <target_column_name> : { <source_column_name> | <expression> }
         [ ... ] ] }
   [METADATA:
      [SCHEMA: <metadata_schema_name>]]
   COMMIT:
      MAX_ROW: <num_rows>
      MINIMAL_INTERVAL: <wait_time>
      CONSISTENCY: { strong | at-least | at-most | none }
   [POLL:
      BATCHSIZE: <num_records>
      TIMEOUT: <poll_time>]
   [PROPERTIES:
      <kafka_property_name>: <kafka_property_value>
      [ ... ]]
[SCHEDULE:
   RETRY_INTERVAL: <retry_time>
   MAX_RETRIES: <num_retries> ]

資料庫相關配置

KAFKA:INPUT配置

KAFKA:INPUT:SOURCE

KAFKA:INPUT:KEY和KAFKA:INPUT:VALUE

Kafka訊息的Key值欄位名稱、資料類型和資料格式。

Kafka訊息的Value欄位名稱、資料類型和資料格式。

必須按照在Key和Value中出現的順序指定所有Kafka資料元素。

KAFKA:INPUT:KEY和KAFKA:INPUT:VALUE至少需要配置一個，如果兩個都未配置會報錯。

KAFKA:INPUT:META

META不是必填項，當您需要展示Message Meta資訊時配置。

KAFKA:INPUT:ERROR_LIMIT

錯誤資料的容忍閾值。當寫入的錯誤資料達到ERROR_LIMIT時，Streaming Server會退出當前任務，自動停止將資料來源的資料寫入雲原生資料倉儲AnalyticDB PostgreSQL版。預設值為0，即Streaming Server會在出現第一次錯誤資料時就退出當前任務，停止資料寫入。ERROR_LIMIT值必須大於1。

目前該參數未啟用，不選擇或者填0即可。

KAFKA:OUTPUT配置

資料庫相關配置

資料寫入到雲原生資料倉儲AnalyticDB PostgreSQL版資料庫的相關配置，包括Kafka值到目標資料庫的映射、寫入模式等。

KAFKA:OUTPUT:MAPPING

KAFKA:METADATA配置

KAFKA:COMMIT配置

COMMIT用於控制向資料庫提交資料的行為。

KAFKA:POLL配置

POLL用於控制Kafka Consumer消費資料的行為。

KAFKA:PROPERTIES配置

PROPERTIES用於配置Kafka Connect，當前採用白名單制，僅支援配置group.id，auto.offset.reset 和isolation.level。詳細資料，請參見Kafka Connect Configs。