全部产品

兼容Kafka

DataHub已经完全兼容Kafka协议,您可以使用原生Kafka客户端对DataHub进行读写操作。

相关介绍

Kafka映射DataHub介绍

Topic类型

Kafka的Topic扩容方式和DataHub的topic扩容方式不同,为了适配Kafka的topic扩容方式,DataHub创建topic时需要将扩容方式选为扩展模式。扩展模式的topic,不再支持分裂/合并操作,而是添加shard的方式,暂不支持减少shard。

Topic命名

Kafka的Topic映射之后为DataHub的project+topic,project和topic以 “.”分割,例如:test_project.test_topic对应到DataHub中Project为test_project,Topic为test_topic,如果含有多个“.”,会以首个“.”分割Project和Topic,多余的“.”和”-“会被替换为“_“。

Partition

DataHub的每个处于Active状态shard对应Kafka的1个Partition,如果当前Active状态shard为5个,那么就可以视为Kafka有5个Partition,写入数据时,可以指定Partition范围为[0,4],如果不指定,则会由kafka客户端选择Partition。

Tuple Topic

Kafka的数据写入Tuple Topic时,Topic Schema必须为2列或1列,类型必须为STRING,其他情况会写入失败。如果为1列,则只写入value,key的数据将被丢弃,如果为2列,则第1列和第2列分别对应key和value。Tuple Topic写入二进制数据会存在乱码问题,二进制数据建议写入Blob Topic

Blob Topic

Kafka的数据写入Blob Topic时,会把Kafka数据的value写入Blob中,如果Kafka数据的key不为NULL,则会写入DataHub的Attribute,其中key为”__kafka_key__“,value为Kafka数据的key。

Header

Kafka的Header对应DataHub的Attribute,但是如果Kafka的Header的value为NULL,则会忽略掉对应的header。建议不要使用”__kafka_key__“作为Header的key

Kafka配置参数

C=Consumer, P=Producer, S=Streams

参数 C/P/S 可选配置 是否必须 描述
bootstrap.servers * 参考Kafka域名列表
security.protocol * SASL_SSL 为了保证数据传输的安全性,Kafka写入DataHub默认使用SSL加密传输
sasl.mechanism * PLAIN AK认证方式,仅支持PLAIN
compression.type P LZ4 是否开启压缩传输,目前仅支持LZ4
group.id C project.topic:subId 必须和订阅的topic保持一致,否则无法读取数据
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 必须和订阅的topic保持一致,否则无法读取数据

以上是使用Kafka客户端写入DataHub需要重点关注的参数,对于等客户端相关的参数,行为没有变化,例如:retries,batch.size;对于服务端相关参数不会对服务端行为有改变,例如:无论acks的值为多少,DataHub默认数据完全写入成功之后才会返回。

Kafka域名列表

地区 Region 外网Endpoint 经典网络ECS Endpoint VPC ECS Endpoint
华东1(杭州) cn-hangzhou dh-cn-hangzhou.aliyuncs.com:9092 dh-cn-hangzhou.aliyun-inc.com:9093 dh-cn-hangzhou-int-vpc.aliyuncs.com:9094
华东2(上海) cn-shanghai dh-cn-shanghai.aliyuncs.com:9092 dh-cn-shanghai.aliyun-inc.com:9093 dh-cn-shanghai-int-vpc.aliyuncs.com:9094
华北2(北京) cn-beijing dh-cn-beijing.aliyuncs.com:9092 dh-cn-beijing.aliyun-inc.com:9093 dh-cn-beijing-int-vpc.aliyuncs.com:9094
华南1(深圳) cn-shenzhen dh-cn-shenzhen.aliyuncs.com:9092 dh-cn-shenzhen.aliyun-inc.com:9093 dh-cn-shenzhen-int-vpc.aliyuncs.com:9094
华北3(张家口) cn-zhangjiakou dh-cn-zhangjiakou.aliyuncs.com:9092 dh-cn-zhangjiakou.aliyun-inc.com:9093 dh-cn-zhangjiakou-int-vpc.aliyuncs.com:9094
亚太东南1(新加坡) ap-southeast-1 dh-ap-southeast-1.aliyuncs.com:9092 dh-ap-southeast-1.aliyun-inc.com:9093 dh-ap-southeast-1-int-vpc.aliyuncs.com:9094
亚太东南3(吉隆坡) ap-southeast-3 dh-ap-southeast-3.aliyuncs.com:9092 dh-ap-southeast-3.aliyun-inc.com:9093 dh-ap-southeast-3-int-vpc.aliyuncs.com:9094
亚太南部1(孟买) ap-south-1 dh-ap-south-1.aliyuncs.com:9092 dh-ap-south-1.aliyun-inc.com:9093 dh-ap-south-1-int-vpc.aliyuncs.com:9094
欧洲中部1(法兰克福) eu-central-1 dh-eu-central-1.aliyuncs.com:9092 dh-eu-central-1.aliyun-inc.com:9093 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.aliyun-inc.com:9093 dh-cn-shanghai-finance-1-int-vpc.aliyuncs.com:9094

示例

创建Topic示例

页面创建

kafka

代码创建

注意:目前无法通过kafka的api创建topic,只能通过datahub的sdk创建,创建时需要指定ExpandMode为ONLY_EXTEND。

maven依赖版本需为2.19.0或更高版本

   
  1. <dependency>
  2. <groupId>com.aliyun.datahub</groupId>
  3. <artifactId>aliyun-sdk-datahub</artifactId>
  4. <version>2.19.0-public</version>
  5. </dependency>
   
  1. public class CreateTopic {
  2. public static void main(String[] args) {
  3. DatahubClient datahubClient = DatahubClientBuilder.newBuilder()
  4. .setDatahubConfig(
  5. new DatahubConfig("https://dh-cn-hangzhou.aliyuncs.com",
  6. new AliyunAccount("accessId", "accessKey")))
  7. .build();
  8. int shardCount = 1;
  9. int lifeCycle = 7;
  10. try {
  11. datahubClient.createTopic("test_project", "test_topic", shardCount, lifeCycle, RecordType.BLOB, "comment", ExpandMode.ONLY_EXTEND);
  12. } catch (DatahubClientException e) {
  13. e.printStackTrace();
  14. }
  15. }
  16. }

Producer示例:

生成kafka_client_producer_jaas.conf文件

创建文件kafka_client_producer_jaas.conf,保存到任意路径,文件内容如下。

   
  1. KafkaClient {
  2. org.apache.kafka.common.security.plain.PlainLoginModule required
  3. username="accessId"
  4. password="accessKey";
  5. };

maven依赖

Kafka-client版本至少大于等于0.10.0.0,推荐2.4.0

   
  1. <dependency>
  2. <groupId>org.apache.kafka</groupId>
  3. <artifactId>kafka-clients</artifactId>
  4. <version>2.4.0</version>
  5. </dependency>

示例代码

   
  1. public class ProducerExample {
  2. static {
  3. System.setProperty("java.security.auth.login.config", "src/main/resources/kafka_client_producer_jaas.conf");
  4. }
  5. public static void main(String[] args) {
  6. Properties properties = new Properties();
  7. properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
  8. properties.put("security.protocol", "SASL_SSL");
  9. properties.put("sasl.mechanism", "PLAIN");
  10. properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  11. properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  12. properties.put("compression.type", "lz4");
  13. String KafkaTopicName = "test_project.test_topic";
  14. Producer<String, String> producer = new KafkaProducer<String, String>(properties);
  15. try {
  16. List<Header> headers = new ArrayList<>();
  17. RecordHeader header1 = new RecordHeader("key1", "value1".getBytes());
  18. RecordHeader header2 = new RecordHeader("key2", "value2".getBytes());
  19. headers.add(header1);
  20. headers.add(header2);
  21. ProducerRecord<String, String> record = new ProducerRecord<>(KafkaTopicName, 0, "key", "Hello DataHub!", headers);
  22. // sync send
  23. producer.send(record).get();
  24. } catch (InterruptedException e) {
  25. e.printStackTrace();
  26. } catch (ExecutionException e) {
  27. e.printStackTrace();
  28. } finally {
  29. producer.close();
  30. }
  31. }
  32. }

运行结果

运行成功之后,可以再DataHub抽样一下,确认是否正常DataHub。

undefined

Consumer示例

生成kafka_client_producer_jaas.conf文件和maven依赖参考Producer示例

示例代码

新加入的consumer需要1分钟左右分配shard,分配完成后即可消费。

   
  1. public class ConsumerExample {
  2. static {
  3. System.setProperty("java.security.auth.login.config", "src/main/resources/kafka_client_producer_jaas.conf");
  4. }
  5. public static void main(String[] args) {
  6. Properties properties = new Properties();
  7. properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
  8. properties.put("security.protocol", "SASL_SSL");
  9. properties.put("sasl.mechanism", "PLAIN");
  10. properties.put("group.id", "test_project.test_topic:1611039998153N71KM");
  11. properties.put("auto.offset.reset", "earliest");
  12. properties.put("session.timeout.ms", "60000");
  13. properties.put("heartbeat.interval.ms", "40000");
  14. properties.put("ssl.endpoint.identification.algorithm", "");
  15. properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  16. properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  17. KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
  18. kafkaConsumer.subscribe(Collections.singletonList("test_project.test_topic"));
  19. while (true) {
  20. ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(5));
  21. for (ConsumerRecord<String, String> record : records) {
  22. System.out.println(record.toString());
  23. }
  24. }
  25. }
  26. }

运行结果

运行成功之后,便可以在终端看到读取到的数据。

   
  1. 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!)

注意:这里同一个请求返回的数据的LogAppendTime是相同的,是该请求返回所有的数据的写入DataHub时间的最大值

Streams示例

maven依赖

   
  1. <dependency>
  2. <groupId>org.apache.kafka</groupId>
  3. <artifactId>kafka-clients</artifactId>
  4. <version>2.4.0</version>
  5. </dependency>
  6. <dependency>
  7. <groupId>org.apache.kafka</groupId>
  8. <artifactId>kafka-streams</artifactId>
  9. <version>2.4.0</version>
  10. </dependency>

代码示例

这里读取test_project下input的数据,将key和value的字符串转为小写重新写入output。

   
  1. public class StreamExample {
  2. static {
  3. System.setProperty("java.security.auth.login.config", "src/main/resources/kafka_client_producer_jaas.conf");
  4. }
  5. public static void main(final String[] args) {
  6. final String input = "test_project.input";
  7. final String output = "test_project.output";
  8. final Properties properties = new Properties();
  9. properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
  10. properties.put("application.id", "test_project.input:1611293595417QH0WL");
  11. properties.put("security.protocol", "SASL_SSL");
  12. properties.put("sasl.mechanism", "PLAIN");
  13. properties.put("session.timeout.ms", "60000");
  14. properties.put("heartbeat.interval.ms", "40000");
  15. properties.put("auto.offset.reset", "earliest");
  16. final StreamsBuilder builder = new StreamsBuilder();
  17. TestMapper testMapper = new TestMapper();
  18. builder.stream(input, Consumed.with(Serdes.String(), Serdes.String()))
  19. .map(testMapper)
  20. .to(output, Produced.with(Serdes.String(), Serdes.String()));
  21. final KafkaStreams streams = new KafkaStreams(builder.build(), properties);
  22. final CountDownLatch latch = new CountDownLatch(1);
  23. Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
  24. @Override
  25. public void run() {
  26. streams.close();
  27. latch.countDown();
  28. }
  29. });
  30. try {
  31. streams.start();
  32. latch.await();
  33. } catch (final Throwable e) {
  34. System.exit(1);
  35. }
  36. System.exit(0);
  37. }
  38. static class TestMapper implements KeyValueMapper<String, String, KeyValue<String, String>> {
  39. @Override
  40. public KeyValue<String, String> apply(String s, String s2) {
  41. return new KeyValue<>(StringUtils.lowerCase(s), StringUtils.lowerCase(s2));
  42. }
  43. }
  44. }

运行结果

启动Streams任务之后,分配shard大概需要1分钟左右,1分钟之后就可以在控制台看到当前的task数量,task数量和输入topic的shard数量保持一致,示例输入topic为3个shard。

   
  1. currently assigned active tasks: [0_0, 0_1, 0_2]
  2. currently assigned standby tasks: []
  3. revoked active tasks: []
  4. revoked standby tasks: []

shard分配成功之后,可以向input中写入一组测试数据 (AAAA,BBBB),(CCCC,DDDD),(EEEE,FFFF),之后再output抽样一下,查看数据是否正确写入。

undefined

注意事项

  • 目前不支持事务、幂等
  • 目前Kafka客户端无法自动创建DataHub Topic,写入之前需要保证已创建Topic
  • Consumer目前最多只可以订阅一个topic
  • Consumer读取的数据时间戳均为LogAppendTime,表示DataHub的落盘时间,单个请求返回的所有数据时间戳相同,为所有数据时间戳的最大值,所以如果读取的时间戳可能会大于实际的落盘时间
  • Streams输入topic目前仅支持一个,输出可以多个topic
  • Streams目前只支持无状态的任务。
  • 支持Kafka版本为0.10.0 -> 2.4.0

常见问题

Q: 写入数据时连接断开

   
  1. Selector - [Producer clientId=producer-1] Connection with dh-cn-shenzhen.aliyuncs.com/120.25.112.216 disconnected
  2. java.io.EOFException
  3. at org.apache.kafka.common.network.SslTransportLayer.read(SslTransportLayer.java:573)
  4. ...

A: Kafka meta请求和写数据请求不是一个连接,第一次meta请求会请求建立一个连接,然后写数据时会重新和meta中的返回的broker重新建立一个连接,并且之后所有的请求都是在第二个连接上发送,因此第一个连接就会闲置,服务端会主动关闭闲置超过一定时间的连接,因此如果这个错误并没有影像数据的正常写入,直接忽略即可。

Q: 启动kafka客户端失败

   
  1. Caused by: org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
  2. Caused by: javax.net.ssl.SSLHandshakeException: No subject alternative names matching IP address 100.67.134.161 found

A: 添加配置properties.put("ssl.endpoint.identification.algorithm", "");

Q: Consumer消费过程中出现DisconnectException

   
  1. [INFO][Consumer clientId=client-id, groupId=consumer-project.topic:subid] Error sending fetch request (sessionId=INVALID, epoch=INITIAL) to node 1: {}.
  2. org.apache.kafka.common.errors.DisconnectException

A: Kafka的客户端需要与服务端保持TCP长连接,一般情况是因为网络抖动造成的,客户端有重试逻辑,因此不会对客户端的消费造成影响。

Q: 使用Kafka客户端访问DataHub时,报错 INVALID_TOPIC_EXCEPTION

A: 一般情况下有两种原因:

  1. DataHub的topic的schema不符合要求(1列或2列的STRING的TUPLE、BLOB)
  2. DataHub的topic不是扩展模式