訪問CDC資料 - - 阿里雲

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。

使用限制

僅支援HBase表（表示通過HBase用戶端寫入Lindorm寬表引擎的表）。
即時資料訂閱功能的資料消費格式僅支援JSON格式檔案。

作業提交方式

您可以通過以下兩種方法編寫並提交Lindorm CDC資料來源的Spark作業。

說明

讀寫Lindorm CDC資料來源的文法說明請參見Lindorm CDC資料來源配置介紹與文法。

Lindorm CDC資料來源配置介紹與文法

Lindorm CDC資料來源的庫表結構介紹

Lindorm計算引擎提供的Lindorm CDC資料來源名稱為lindorm_cdc。
Lindorm CDC資料來源不支援管理Namespace，僅支援管理表。表名稱為建立資料訂閱通道的主題名。

Lindorm CDC資料來源輸出的Schema介紹

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資料來源的配置項介紹

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資料來源的文法介紹

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`對應的欄位，其他欄位為空白。