DataHub Kafka兼容模式

DataHub Kafka兼容模式

DataHub 已经兼容 Kafka 的协议，Kafka 可以直接使用 Kafka 的 sdk 来连接 DataHub 服务，进行数据的订阅和发布。

DataHub&Kafka概念映射

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。

目前单个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。

快速开始

资源创建

首先登录DataHub 控制台，创建 project

创建 topic

创建 group 并绑定需要消费的 topic，创建完成以后也可以修改绑定的 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抽样一下，查看数据是否正确写入。

自建Kafka切流到DataHub

1. 切换broker地址，具体可以参考下文的域名列表；

2. 在DataHub上创建资源并修改代码中的资源名称，具体可以参考上文的资源创建；

3. 认证方式切换为SASL_SSL，具体使用的认证机制是PLAIN，username和password分别填写阿里云的ak和sk即可；

附录

配置介绍

C=Consumer, P=Producer, S=Streams

服务域名列表

Kafka API 兼容情况

Kafka 官方文档列出了所有的 api，为了方便用户了解 kod 的兼容情况，我们也把已兼容的 API 列表给出。