创建数据订阅-云原生多模数据库 Lindorm(Lindorm)-阿里云帮助中心

前提条件

已将客户端IP添加至白名单中，具体操作请参见设置白名单。
已开通数据订阅功能，具体操作，请参见开通数据订阅。
确保源实例、目标实例和LTS服务网络互通，详细信息请参见网络互通。
已添加Lindorm宽表数据源，具体操作请参见Lindorm宽表数据源。
已添加Kafka数据源，具体操作请参见Kafka数据源。

通过Pull模式创建数据订阅通道

进入LTS（原BDS）页面，在左侧导航栏中，选择数据订阅 > Pull模式。

进入 Lindorm CDC 的 订阅通道列表 页面，页面右上角提供 创建数据订阅通道 按钮，列表展示已有通道的订阅通道 Id、Lindorm 表名、主题名及操作（详情、删除）。

单击创建数据订阅通道，并配置以下参数。

名称	描述
源集群	填写Lindorm实例ID。
Lindorm表名	选择需要创建数据订阅通道的Lindorm实例表，一条通道只能选择一张表格。
主题名	用于消费数据的主题名称。
数据过期时间（天）	表示数据可以保存的天数，默认为7天。
主题分区数	表示Kafka客户端为该主题设置多个分区，多分区可以并发消费数据，默认为4个分区。

单击提交。
（可选）找到目标订阅通道，单击操作列的详情，可以查看数据订阅通道详情、消费详情和存储详情。

（可选）您可以通过以下代码在Kafka客户端对订阅数据进行消费。

import org.apache.hadoop.hbase.util.Bytes;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
public class TestConsume {
  public static void main(String[] args) throws Exception {
    // 创建订阅通道时填写的topic名称
    String topic = "test-topic";
    // 链接endpoint的配置项
    Properties props = new Properties();
    // 指定链接endpoint地址
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "ld-xxx:9092");
    // 指定Key序列化器，不可更改
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
    // 指定Value序列化器，不可更改
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
    // 指定消费组名称，在消费时会自动创建
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id-0");
    // 创建消费者
    KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props);
    // 订阅主题
    consumer.subscribe(Arrays.asList(topic));
    // 用消费者拉取数据
    ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(10000));
    for (ConsumerRecord<byte[], byte[]> record : records) {
      // 查看数据内容
      System.out.println("key: " + Bytes.toString(record.key()));
      System.out.println("value: " + Bytes.toString(record.value()));
    }
    // 提交当前消费位移
    consumer.commitSync();
    // 关闭消费者
    consumer.close();
  }
}

说明

数据消费格式说明请参见数据消费格式。

通过Push模式创建数据订阅

推送流程

数据订阅功能将云原生多模数据库 Lindorm表格的增量数据消息主动推送至下游Kafka，推送流程如下。

创建Lindorm Streams

进入LTS（原BDS）页面，在左侧导航栏中，选择数据订阅 > Push模式。

单击create，并配置以下参数。

参数	描述
Lindorm集群	选择已添加Lindorm宽表数据源。
表名	需要填写数据订阅的表名，格式为：namespace.tablename。例如：ns1.table1表示订阅空间名称为ns1中的表table1数据。
黑名单表（可选）	填写黑名单表名。该表中的增量数据不会被推送。
MessageStorage Type	选择KAFKA。
Storage Datasource	选择已添加Kafka数据源。
MessageStorage Config	kafka_topic表示需要填写Kafka Topic的名称。 kafka_ttl：无需填写，请忽略。 kafka_partition_num：无需填写，请忽略。重要 Push模式下Topic需要在Kafka中提前建好。
MessageVersion	消费格式，默认格式为DebeziumV2。
Message Config	`old_image`表示消息中是否包含当前行修改前的值，需填写为`true`，无需填写`false`。 `new_image`表示消息中是否包含修改后整行的值，需填写为`true`，无需填写`false`。 `with_schema`表示消息中是否携带表的Schema信息，建议填写为`false`，避免Schema造成消息膨胀。 `ignore_family_prefix`表示导出的列名是否去掉列族，例如完整列名为`f:name`，将该参数配置为`true`后导出的列名为`name`。

单击提交。