DataHub：DataHub Kafka相容模式

DataHub Kafka相容模式

DataHub 已經相容 Kafka 的協議，使用者可以直接使用Kafka的sdk來串連 DataHub 服務,進行資料的訂閱和發布。

DataHub&Kafka概念映射

Kafka	Datahub
Topic	Project.Topic
partition	shard
offset	sequence

Kafka Topic

DataHub 也有 Topic，並且資源層級上 DataHub 還有 Project，而 Kafka 是沒有 Project 的，所以為了相容這個邏輯，DataHub 的 Project 和 Topic 組合後就是 Kafka 的 topic，組合方式採用"."來串連。例如：在 DataHub 中我有 Project 和 Topic 分別是 test_project 和 test_topic，那麼使用 Kafka 協議的時候，Topic 名稱就是 test_project.test_topic。

Kafka Parittion

Kafka 的 partition 和 DataHub 的 shard 一一對應，意義也相同，都是代表一個有序的隊列。

Kafka Consumer Group

DataHub group行為和 Kafka 基本保持一致，使用者可以在project下建立group並綁定想要訂閱的topic，就可以使用該group訂閱這個project下的多個topic。如果group綁定了topic，使用者可以在topic頁面的訂閱列表頁面，看到由group自動建立的訂閱，刪除該訂閱會導致group無法訂閱該topic，並且之前的消費點位都會消失。

因為 group 是 project 的子資源，所以指定時需要和 project 一起指定。例如：在 DataHub 中有 project 和 group 分別是 test_project 和 test_group，那麼使用 Kafka 協議的時候，group 名稱就是 test_project.test_group。

在建立消費組對話方塊中，填寫名稱和描述，在Topic地區勾選需要訂閱的 Topic，單擊 > 按鈕將其添加至右側已選列表，然後單擊建立。

目前單個group限制最多可以訂閱50個topic，如果需要訂閱更多，請開工單聯絡我們。

Kafka Record

Kafka Record 是 key-val 的形勢，而 DataHub 的 Record 分為兩種形式 Tuple 和 Blob，Tuple 是強結構化資料， BlobRecord 是位元據。對於Kafka Record 一般分為兩部分資訊，Header 和 key-value。

Kafka 的 header

Kafka header 可以給資料添加額外資訊，這個和 DataHub 的 attribute 作用一致，如果使用 Kafka 用戶端寫入資料時攜帶了 header 資訊，那麼這個資訊會放到 DataHub 的 attribute 上。如果Kafka的Header的value為NULL，則會忽略掉對應的header。建議不要使用"__kafka_key__"作為Header的key，這個是內部保留 key。

Kafka 的 key-value 資料

• DataHub Topic 是 Tuple 類型，key 對應第一列 String，value 對應第二列 String

• DataHub Topic 是 Blob 類型，value 對應資料內容，key 放入 attribute，其中格式為<"__kafka_key__", key>

Kafka Offset

Kafka offset 是一個 64 位元的整型數字，對於一個 partition 來說，offset 從 0 開始自增，因此每條資料都有獨一無二的 offset，主要用於消費點位的記錄。DataHub 的 sequence 和 Kafka 的 offset 含義一致，因此使用者可以直接認為 DataHub 的 sequence 和 Kafka 的 offset 是完全對標的。

使用限制

DataHub不支援Kafka事務、等冪 、SchemaRegistry、Log Compaction。

快速開始

1. 使用kafka client 推薦版本2.4.0，相容0.10.0～4.0。

2. 在 DataHub 上建立對應的資源，Project、Topic 和 Group。

3. 安全認證方式切換為SASL+SSL，SASL 使用的是 PLAIN，同時配置好阿里雲的 AK/SK。

4. Kafka 的broker 資訊切換為 DataHub 的 Endpoint（參考服務網域名稱列表）

資源建立

首先登入DataHub 控制台，建立 project

在建立專案對話方塊中，設定名稱為test_kafka_project，描述為test kafka。

建立 topic

在DataHub控制台的建立Topic頁面，選擇建立方式為直接建立，填寫名稱為test_kafka，類型選擇TUPLE。在Schema詳情中添加兩個欄位：欄位名key和value，類型均為STRING，勾選允許為NULL。設定Shard數量為1，生命週期為3，開啟Shard擴充模式，關閉啟動多Version。填寫描述後單擊建立。

建立 group 並綁定需要消費的 topic，建立完成以後也可以修改綁定的 topic 列表。如果是非消費情境，那麼可以直接跳過這一步。

在建立Group面板中，填寫名稱和描述，通過穿梭框將需要消費的 Topic 從左側可選列表移至右側已選列表，然後單擊建立。

認證方式

建立檔案kafka_client_producer_jaas.conf，儲存到任意路徑，檔案內容如下。

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="accessId"
  password="accessKey";
};

同時配置中添加如下內容

// 把ak的設定檔目錄添加進來，也可以通過設定環境變數的方式設定
static {
        System.setProperty("java.security.auth.login.config", "/xxxpath/kafka_client_producer_jaas.conf");
    }
// 安全認證方式設定為 SASL + SSL
// Properties properties = new Properties();
properties.put("security.protocol", "SASL_SSL");
properties.put("sasl.mechanism", "PLAIN");

Producer 樣本

kafka clients pom 檔案

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>2.4.0</version>
</dependency>

ProducerExample

public class ProducerExample {
    static {
        System.setProperty("java.security.auth.login.config", "/path/xxx/kafka_client_producer_jaas.conf");
    }
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("compression.type", "lz4");
        properties.put("enable.idempotence", "false"); // 3.0.1及以上版本需要添加
        String KafkaTopicName = "test_project.test_topic";
        Producer<String, String> producer = new KafkaProducer<String, String>(properties);
        try {
            List<Header> headers = new ArrayList<>();
            RecordHeader header1 = new RecordHeader("key1", "value1".getBytes());
            RecordHeader header2 = new RecordHeader("key2", "value2".getBytes());
            headers.add(header1);
            headers.add(header2);
            ProducerRecord<String, String> record = new ProducerRecord<>(KafkaTopicName, 0, "key", "Hello DataHub!", headers);
            // sync send
            producer.send(record).get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}

運行成功之後，可以在 DataHub 控制台進行資料抽樣查看，確認資料是否正常寫入 DataHub。

Consumer 樣本

ConsumerExample

package com.aliyun.datahub.kafka.demo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class ConsumerExample2 {
    static {
        System.setProperty("java.security.auth.login.config", "/path/xxx/kafka_client_producer_jaas.conf");
    }
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("group.id", "test_project.test_kafka_group");
        properties.put("auto.offset.reset", "earliest");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("ssl.endpoint.identification.algorithm", "");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
        List<String> topicList = new ArrayList<>();
        topicList.add("test_project.test_topic1");
        topicList.add("test_project.test_topic2");
        topicList.add("test_project.test_topic3");
        kafkaConsumer.subscribe(topicList);
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(5));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.toString());
            }
        }
    }
}

運行成功之後，便可以在終端看到讀取到的資料。

ConsumerRecord(topic = test_project.test_topic, partition = 0, leaderEpoch = 0, offset = 0, LogAppendTime = 1611040892661, serialized key size = 3, serialized value size = 14, headers = RecordHeaders(headers = [RecordHeader(key = key1, value = [118, 97, 108, 117, 101, 49]), RecordHeader(key = key2, value = [118, 97, 108, 117, 101, 50])], isReadOnly = false), key = key, value = Hello DataHub!)

Streams樣本

這裡讀取test_project下input的資料，將key和value的字串轉為小寫重新寫入output。

public class StreamExample {
    static {
        System.setProperty("java.security.auth.login.config", "/path/xxx/kafka_client_producer_jaas.conf");
    }
    public static void main(final String[] args) {
        final String input = "test_project.input";
        final String output = "test_project.output";
        final Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("application.id", "test_project.test_kafka_group");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("auto.offset.reset", "earliest");
        final StreamsBuilder builder = new StreamsBuilder();
        TestMapper testMapper = new TestMapper();
        builder.stream(input, Consumed.with(Serdes.String(), Serdes.String()))
                .map(testMapper)
                .to(output, Produced.with(Serdes.String(), Serdes.String()));
        final KafkaStreams streams = new KafkaStreams(builder.build(), properties);
        final CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });
        try {
            streams.start();
            latch.await();
        } catch (final Throwable e) {
            System.exit(1);
        }
        System.exit(0);
    }
    static class TestMapper implements KeyValueMapper<String, String, KeyValue<String, String>> {
        @Override
        public KeyValue<String, String> apply(String s, String s2) {
            return new KeyValue<>(StringUtils.lowerCase(s), StringUtils.lowerCase(s2));
        }
    }
}

啟動Streams任務之後，分配shard大概需要1分鐘左右，1分鐘之後就可以在控制台看到當前的task數量，task數量和輸入topic的shard數量保持一致，樣本輸入topic為3個shard。

currently assigned active tasks: [0_0, 0_1, 0_2]
    currently assigned standby tasks: []
    revoked active tasks: []
    revoked standby tasks: []

shard分配成功之後，可以向input中寫入一組測試資料 (AAAA,BBBB),(CCCC,DDDD),(EEEE,FFFF)，之後再output抽樣一下，查看資料是否正確寫入。

output 抽樣結果顯示資料已正確寫入，經 Streams 任務的 TestMapper 處理後，大寫輸入被轉換為小寫輸出：key 分別為 aaaa、cccc、eeee，對應 val 分別為 bbbb、dddd、ffff。

自建Kafka切流到DataHub

1. 切換broker地址，具體可以參考下文的網域名稱列表；

2. 在DataHub上建立資源並修改代碼中的資源名稱，具體可以參考上文的資源建立；

3. 認證方式切換為SASL_SSL，具體使用的認證機制是PLAIN，username和password分別填寫阿里雲的ak和sk即可；

附錄

配置介紹

C=Consumer, P=Producer, S=Streams

參數	C/P/S	可選配置	是否必須	描述
bootstrap.servers	*	參考網域名稱列表	是
security.protocol	*	SASL_SSL	是	為了保證資料轉送的安全性，Kafka寫入DataHub預設使用SSL加密傳輸
sasl.mechanism	*	PLAIN	是	AK認證方式，僅支援PLAIN
compression.type	P	LZ4	否	是否開啟壓縮傳輸，目前僅支援LZ4
enable.idempotence	P	false	否	kafka client 從 3.0.1 開始預設開啟了等冪，因 DataHub 暫不支援等冪，所以需要手動關閉，小於 3.0.1 的版本不需要添加此配置
group.id	C	project.group	是
partition.assignment.strategy	C	org.apache.kafka.clients.consumer.RangeAssignor	否	Kafka預設為RangeAssignor，並且DataHub目前只支援RangeAssignor，請不要修改此配置
session.timeout.ms	C/S	[60000, 180000]	否	kafka預設為10000, 但是因為DataHub限制最小為60000,所以這裡預設會變為60000
heartbeat.interval.ms	C/S	建議session.timeout.ms的 2/3	否	Kafka預設為3000，但是因為session.timeout.ms會被預設修改為60000，所以這裡建議顯示設定為40000，否則heartbeat請求會過於頻繁
application.id	S	project.topic:subId或者project.group	是	使用project.topic:subId時必須和訂閱的topic保持一致，否則無法讀取資料，推薦使用project.group

服務網域名稱列表

地區	Region	公網Endpoint	VPC ECS Endpoint
華東1(杭州)	cn-hangzhou	dh-cn-hangzhou.aliyuncs.com:9092	dh-cn-hangzhou-int-vpc.aliyuncs.com:9094
華東2(上海)	cn-shanghai	dh-cn-shanghai.aliyuncs.com:9092	dh-cn-shanghai-int-vpc.aliyuncs.com:9094
華北2(北京)	cn-beijing	dh-cn-beijing.aliyuncs.com:9092	dh-cn-beijing-int-vpc.aliyuncs.com:9094
烏蘭察布	cn-wulanchabu	dh-cn-wulanchabu.aliyuncs.com:9092	dh-cn-wulanchabu-int-vpc.aliyuncs.com:9094
華南1(深圳)	cn-shenzhen	dh-cn-shenzhen.aliyuncs.com:9092	dh-cn-shenzhen-int-vpc.aliyuncs.com:9094
華北3(張家口)	cn-zhangjiakou	dh-cn-zhangjiakou.aliyuncs.com:9092	dh-cn-zhangjiakou-int-vpc.aliyuncs.com:9094
亞太地區東南1(新加坡)	ap-southeast-1	dh-ap-southeast-1.aliyuncs.com:9092	dh-ap-southeast-1-int-vpc.aliyuncs.com:9094
亞太地區東南3(吉隆坡)	ap-southeast-3	dh-ap-southeast-3.aliyuncs.com:9092	dh-ap-southeast-3-int-vpc.aliyuncs.com:9094
歐洲中部1(法蘭克福)	eu-central-1	dh-eu-central-1.aliyuncs.com:9092	dh-eu-central-1-int-vpc.aliyuncs.com:9094
上海金融雲	cn-shanghai-finance-1	dh-cn-shanghai-finance-1.aliyuncs.com:9092	dh-cn-shanghai-finance-1-int-vpc.aliyuncs.com:9094
中國香港	cn-hongkong	dh-cn-hongkong.aliyuncs.com:9092	dh-cn-hongkong-int-vpc.aliyuncs.com:9094

Kafka API 相容情況

Kafka 官方文檔列出了所有的 api，為了方便使用者瞭解 kod 的相容情況，我們也把已相容的 API 列表給出。

介面	描述
Produce	資料寫入
Fetch	資料讀取
ListOffsets	舊版本根據時間擷取offset列表，新版本根據時間擷取offset
Metadata	讀寫資料時擷取meta資訊
OffsetCommit	提交點位
OffsetFetch	擷取點位
FindCoordinator	尋找coordinator所在broker，直接返回vip即可
JoinGroup	加入group
Heartbeat	心跳
LeaveGroup	離開group
SyncGroup	leader用來發送分配方案，leader和follower擷取分配方案
SaslHandshake	鑒權相關
ApiVersions	擷取所有可用介面
CreateTopics	建立topic
DeleteTopics	刪除topic
OffsetForLeaderEpoch	擷取最新的offset
SaslAuthenticate	鑒權相關
CreatePartitions	添加partition
DeleteGroups	刪除group
OffsetDelete	清除點位