Kafka數據遷移MaxCompute最佳實踐

摘要： 本文向您詳細介紹如何使用DataWorks數據同步功能，將Kafka集羣上的數據遷移到阿里雲MaxCompute大數據計算服務。

前提條件

搭建Kafka集羣

進行數據遷移前，您須要保證本身的Kafka集羣環境正常。本文使用阿里雲EMR服務自動化搭建Kafka集羣，詳細過程請參見：Kafka 快速入門。

本文使用的EMR Kafka版本信息以下：
EMR版本: EMR-3.12.1
集羣類型: Kafka
軟件信息: Ganglia 3.7.2 ZooKeeper 3.4.12 Kafka 2.11-1.0.1 Kafka-Manager 1.3.3.16
Kafka集羣使用專有網絡，區域爲華東1（杭州），主實例組ECS計算資源配置公網及內網IP，具體配置以下圖所示。

建立MaxCompute 項目

開通MaxCompute服務並建立好項目，本文中在華東1（杭州）區域建立項目bigdata_DOC，同時啓動DataWorks相關服務，以下圖所示。詳情請參見開通MaxCompute。

背景信息

Kafka是一款分佈式發佈與訂閱的消息中間件，具備高性能、高吞量的特色被普遍使用，每秒能處理上百萬的消息。Kafka適用於流式數據處理，主要應用於用戶行爲跟蹤、日誌收集等場景。

一個典型的Kafka集羣包含若干個生產者（Producer）、Broker、消費者（Consumer）以及一個Zookeeper集羣。Kafka集羣經過Zookeeper管理自身集羣的配置並進行服務協同。

Topic是Kafka集羣上最經常使用的消息的集合，是一個消息存儲邏輯概念。物理磁盤不存儲Topic，而是將Topic中具體的消息按分區（Partition）存儲在集羣中各個節點的磁盤上。每一個Topic能夠有多個生產者向它發送消息，也能夠有多個消費者向它拉取（消費）消息。

每一個消息被添加到分區時，會分配一個offset（偏移量，從0開始編號），是消息在一個分區中的惟一編號。

操做步驟

準備測試表與數據
Kafka集羣建立測試數據
爲保證您能夠順利登錄EMR集羣Header主機及MaxCompute和DataWorks能夠順利和EMR集羣Header主機通訊，請您首先配置EMR集羣Header主機安全組，放通TCP 22及TCP 9092端口。
登陸EMR集羣Header主機地址
進入EMR Hadoop控制檯集羣管理 > 主機列表頁面，確認EMR集羣Header主機地址，並經過SSH鏈接遠程登陸。

建立測試Topic
使用kafka-topics.sh --zookeeper emr-header-1:2181/kafka-1.0.1 --partitions 10 --replication-factor 3 --topic testkafka --create命令建立測試所使用的Topic testkafka。您可使用kafka-topics.sh --list --zookeeper emr-header-1:2181/kafka-1.0.1命令查看已建立的Topic。
[root@emr-header-1 ~]# kafka-topics.sh --zookeeper emr-header-1:2181/kafka-1.0.1 --partitions 10 --replication-factor 3 --topic testkafka --create
Created topic "testkafka".
[root@emr-header-1 ~]# kafka-topics.sh --list --zookeeper emr-header-1:2181/kafka-1.0.1
__consumer_offsets
_emr-client-metrics
_schemas
connect-configs
connect-offsets
connect-status
testkafka
寫入測試數據
您可使用kafka-console-producer.sh --broker-list emr-header-1:9092 --topic testkafka命令模擬生產者向Topic testkafka中寫入數據。因爲Kafka用於處理流式數據，您能夠持續不斷的向其中寫入數據。 爲保證測試結果，建議您寫入10條以上的數據。
[root@emr-header-1 ~]# kafka-console-producer.sh --broker-list emr-header-1:9092 --topic testkafka

123
abc

爲驗證寫入數據生效，您能夠同時再打開一個SSH窗口，使用kafka-console-consumer.sh --bootstrap-server emr-header-1:9092 --topic testkafka --from-beginning命令模擬消費者，覈驗數據是否已成功寫入Kafka。 以下所示，當數據寫入成功時，您能夠看到已寫入的數據。

[root@emr-header-1 ~]# kafka-console-consumer.sh --bootstrap-server emr-header-1:9092 --topic testkafka --from-beginning
123
abc
建立MaxCompute表
爲保證MaxCompute能夠順利接收Kafka數據，請您首先在MaxCompute上建立表。本例中爲測試便利，使用非分區表。
登錄DataWorks建立表，詳情請參見表管理。

您能夠單擊DDL模式進行建表，建表語句舉例以下。
CREATE TABLE testkafka (

`key` string,
`value` string,
`partition1` string,
`timestamp1` string,
`offset` string,
`t123` string,
`event_id` string,
`tag` string

) ;
其中的每一列，對應於DataWorks數據集成Kafka Reader的默認列，您能夠自主命名。詳情請參見配置Kafka Reader：
__key__表示消息的key。
__value__表示消息的完整內容 。
__partition__表示當前消息所在分區。
__headers__表示當前消息headers信息。
__offset__表示當前消息的偏移量。
__timestamp__表示當前消息的時間戳。
數據同步
新建自定義資源組
因爲當前DataWorks的默認資源組沒法完美支持Kafka插件，您須要使用自定義資源組完成數據同步。自定義資源組詳情請參見新增任務資源。

在本文中，爲節省資源，直接使用EMR集羣Header主機做爲自定義資源組。完成後，請等待服務器狀態變爲可用。

新建並運行同步任務
在您的業務流程中右鍵單擊數據集成，選擇新建數據集成節點 > 數據同步。

新建數據同步節點後，您須要選擇數據來源的數據源爲Kafka，數據去向的數據源爲ODPS，而且使用默認數據源odps_first。選擇數據去向表爲您新建的testkafka。完成上述配置後，請點擊下圖框中的按鈕，轉換爲腳本模式。

腳本配置以下，代碼釋義請參見配置Kafka Reader。
{

"type": "job",
"steps": [
    {
        "stepType": "kafka",
        "parameter": {
            "server": "47.xxx.xxx.xxx:9092",
            "kafkaConfig": {
                "group.id": "console-consumer-83505"
            },
            "valueType": "ByteArray",
            "column": [
                "__key__",
                "__value__",
                "__partition__",
                "__timestamp__",
                "__offset__",
                "'123'",
                "event_id",
                "tag.desc"
            ],
            "topic": "testkafka",
            "keyType": "ByteArray",
            "waitTime": "10",
            "beginOffset": "0",
            "endOffset": "3"
        },
        "name": "Reader",
        "category": "reader"
    },
    {
        "stepType": "odps",
        "parameter": {
            "partition": "",
            "truncate": true,
            "compress": false,
            "datasource": "odps_first",
            "column": [
                "key",
                "value",
                "partition1",
                "timestamp1",
                "offset",
                "t123",
                "event_id",
                "tag"
            ],
            "emptyAsNull": false,
            "table": "testkafka"
        },
        "name": "Writer",
        "category": "writer"
    }
],
"version": "2.0",
"order": {
    "hops": [
        {
            "from": "Reader",
            "to": "Writer"
        }
    ]
},
"setting": {
    "errorLimit": {
        "record": ""
    },
    "speed": {
        "throttle": false,
        "concurrent": 1,
        "dmu": 1
    }
}

}
您能夠經過在Header主機上使用kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --list命令查看group.id參數，及消費者的Group名稱。
[root@emr-header-1 ~]# kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --list
Note: This will not show information about old Zookeeper-based consumers.

_emr-client-metrics-handler-group
console-consumer-69493
console-consumer-83505
console-consumer-21030
console-consumer-45322
console-consumer-14773
以console-consumer-83505爲例，您能夠根據該參數在Header主機上使用kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --describe --group console-consumer-83505命令確認beginOffset及endOffset參數。
[root@emr-header-1 ~]# kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --describe --group console-consumer-83505
Note: This will not show information about old Zookeeper-based consumers.
Consumer group 'console-consumer-83505' has no active members.
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
testkafka 6 0 0 0 - - -
test 6 3 3 0 - - -
testkafka 0 0 0 0 - - -
testkafka 1 1 1 0 - - -
testkafka 5 0 0 0 - - -
完成腳本配置後，請首先切換任務資源組爲您剛建立的資源組，而後點擊運行。

完成運行後，您能夠在運行日誌中查看運行結果，以下爲成功運行的日誌。

結果驗證

您能夠經過新建一個數據開發任務運行SQL語句，查看當前表中是否已存在從Kafka同步過來的數據。本例中使用select * from testkafka；語句，完成後點擊運行便可。

執行結果以下，本例中爲保證結果，在testkafka Topic中輸入了多條數據，您能夠查驗是否和您輸入的數據一致。

 

本文做者：付帥

原文連接

本文爲雲棲社區原創內容，未經容許不得轉載。