CDC数据源的配置与语法介绍_云原生多模数据库 Lindorm(Lindorm)-阿里云帮助中心

云原生多模数据库 Lindorm计算引擎提供Lindorm CDC（Change Data Capture）数据源，您可以通过该数据源使用Spark SQL查询Lindorm实例其他引擎变更的数据。

前提条件

Lindorm实例已开通LTS（Lindorm Tunnel Service）服务，具体请参见购买并登录LTS。
Lindorm宽表已创建数据订阅通道，创建方法请参见通过Pull模式创建数据订阅通道。
说明
创建数据订阅通道时需要注意以下几点：
- 不要勾选在投递的数据内容中为列名省略family前缀。
- 数据序列化格式中请选择json。
- 一个主题名仅对应一个Lindorm表名。
为HBase表设置LINDORM_HBASE_CATALOG属性，设置方法请参见访问宽表数据。
说明
LINDORM_HBASE_CATALOG属性表示SparkSQL Schema与HBase表Schema的对应关系，Lindorm CDC数据源根据该属性的内容提取HBase表Schema。

您可以通过以下两种方法编写并提交Lindorm CDC数据源的Spark作业。

说明

读写Lindorm CDC数据源的语法说明请参见Lindorm CDC数据源配置介绍与语法。

Lindorm CDC数据源根据LINDORM_HBASE_CATALOG属性提取HBase表的Schema作为Lindorm CDC数据源的Schema。Lindorm CDC数据源从Kafka中读取数据，每一条操作记录都会被保存。Lindorm CDC数据源的Schema支持以下meta字段：

字段名	类型	说明	字段配置值
_cdc_timestamp_kafka	long	该操作记录写入Kafka的时间戳。单位为毫秒。	无需配置，Schema中包含默认配置值。
_cdc_operation_type	string	该操作记录的变更类型。 C：新增数据操作。 U：更新数据操作。 D：删除数据操作。	无需配置，Schema中包含默认配置值。
_cdc_timestamp_lindorm	long	该操作记录被Lindorm引擎处理的时间戳。单位为毫秒。	spark.sql.catalog.lindorm_cdc.lindormTsEnabled
_cdc_timestamp_lts	long	该操作记录被LTS处理的时间戳。单位为毫秒。	spark.sql.catalog.lindorm_cdc.ltsTsEnabled

Lindorm CDC数据源的配置项如下表所示：

配置项	是否必填	说明	示例值
spark.sql.catalog.lindorm_cdc.username	必填：提交JAR作业或者Python作业。非必填（系统自动添加）：提交SQL作业。	连接Lindorm宽表引擎的用户名。	默认用户名root。
spark.sql.catalog.lindorm_cdc.password	必填：提交JAR作业或者Python作业。非必填（系统自动添加）：提交SQL作业。	连接Lindorm宽表引擎的密码。	默认密码root。
spark.sql.catalog.lindorm_cdc.lindormTsEnabled	非必填	Lindorm处理该操作记录的时间戳。默认值为false。设置为true时Lindorm CDC数据源的Schema中会增加_cdc_timestamp_lindorm字段。	true
spark.sql.catalog.lindorm_cdc.ltsTsEnabled	非必填	LTS处理该操作记录的时间戳。默认值为false。设置为true时Lindorm CDC数据源的Schema中会增加_cdc_timestamp_lts字段。	true

Lindorm CDC数据源支持的语法如下表所示：

语法	描述	示例
USE table_name	使用某个表。	USE test
SHOW TABLES	查看所有的表。	SHOW TABLES
DESCRIBE table_name	查看某个表的详细信息。	DESC test或者DESCRIBE test
SELECT	关于SELECT语法请参见Spark SQL。说明 SELECT语法使用中需要注意以下两点：必须使用`_cdc_timestamp_kafka > $startTimestamp and _cdc_timestamp_kafka < $endTimestamp`指定Kafka中的数据读取范围。对于`_cdc_operation_type=D`的字段，仅显示`rowKey`对应的字段，其他字段为空。