tencent cloud

Stream Compute Service

문서Stream Compute ServiceStream Lake Engine SetatsJob Development GuideDeveloping Flink JobsFlink CDC Stream Reading

Flink CDC Stream Reading

Download
포커스 모드
폰트 크기
마지막 업데이트 시간: 2026-07-03 15:10:13

Overview

Setats supports streaming Changelog data from tables in a CDC manner through Flink. After data is written into Setats, the system merges the results based on primary keys to generate change logs. Downstream applications can then continuously consume the new, updated, and deleted events via the setats-cdc connector.
This capability applies to the following scenarios:
Subscribe to primary key change data in Setats in real time.
Use Setats as a unified incremental data source and distribute it to downstream Flink jobs.
Build dimension table caches, perform real-time aggregation, or establish message distribution pipelines based on Changelog.

Prerequisites

You have created a Setats cluster and completed the Warehouse configuration.
You have obtained the manager-url.
An upstream task is continuously writing data to the target Setats table.

Table Creation Syntax

CREATE TABLE `setats_cdc_source` (
  `id` STRING,
  `dt` STRING,
  `name` STRING,
  `address` STRING,
  PRIMARY KEY (`id`, `dt`) NOT ENFORCED
) WITH (
  'connector' = 'setats-cdc',
  'warehouse' = 'cosn://<bucket>/warehouse',
  'catalog-type' = 'hive',
  'catalog-name' = 'setats',
  'catalog-database' = 'testdb',
  'catalog-table' = 'demo_setats_table1',
  'uri' = 'thrift://<metastore-host-1>:7004,thrift://<metastore-host-2>:7004',
  'manager-url' = '<Manager Url>',
  'bucket.discovery-interval' = '30s',
  'startup.mode' = 'earliest'
);


Key Parameter Descriptions

Parameter
Required
Description
Default Value
connector
Yes
Fixed as setats.
(Not provided, must be specified)
manager-url
Yes
Setats Manager address
(Not provided, must be specified)
warehouse
Yes
Setats Warehouse path
(Not provided, must be specified)
catalog-type
Yes
Catalog type, commonly Hive or Hadoop.
(Not provided, must be specified)
catalog-name
Yes
Catalog name
(Not provided, must be specified)
catalog-database
Yes
Database name.
default
catalog-table
Yes
Target table name, which must match the actual table name.
None
uri
Required for Hive catalog.
Hive metastore address
(Not provided, must be specified for Hive catalog)
bucket
Yes
Number of buckets for the target table. It is recommended to configure this based on data volume and concurrent write capability.
(Not provided, must be specified)
bucket-key
Yes
Bucket key, which determines which bucket the data is written to. It is recommended to select a field with relatively uniform distribution and stability. Typically, the primary key field or a subset of the primary key can be used.
Primary key by default. It is recommended to explicitly set a suitable business field as the bucket-key.
bucket.discovery-interval
No
Dynamically detect newly added BucketShadingGroups.
300s
startup.mode
No
latest: read from the latest.
earliest: read from the earliest.
from-timestamp: read from a specified timestamp.
Scan.timestamp-millis: the millisecond timestamp of the startup time in the "timestamp" startup mode.
latest
scan.timestamp-millis
No
In the scan.timestamp-millis mode, set the startup timestamp. Example: 1767196800000.
None

Usage Examples

-- Create a Setats CDC source table
CREATE TABLE `demo_setats_table1` (
  `id` STRING,
  `dt` STRING,
  `name` STRING,
  `address` STRING,
  PRIMARY KEY (`id`, `dt`) NOT ENFORCED
) WITH (
  'connector' = 'setats-cdc',
  'warehouse' = 'cosn://<bucket>/warehouse',
  'catalog-name' = 'setats',
  'catalog-type' = 'hive',
  'uri' = 'thrift://<metastore-host-1>:7004,thrift://<metastore-host-2>:7004',
  'catalog-database' = 'testdb',
  'catalog-table' = 'demo_setats_table1',
  'manager-url' = '<Manager Url>',
  'bucket.discovery-interval' = '30s',
  'startup.mode' = 'earliest'
);
-- Create a debug Sink table
CREATE TABLE `logger_sink_table` (
  `id` STRING,
  `dt` STRING,
  `name` STRING,
  `address` STRING
) WITH (
  'connector' = 'logger',
  'all-changelog-mode' = 'true',
  'print-identifier' = 'DebugData'
);
-- Perform CDC reading
INSERT INTO `logger_sink_table`
SELECT `id`, `dt`, `name`, `address`
FROM `demo_setats_table1`;


Viewing Consumption Results

The logger connector prints CDC results to the TaskManager logs. After you deploy the job, you can view the corresponding logs in the Flink UI / Oceanus console to confirm whether change events are being continuously received.

Notes

The setats-cdc connector reads Changelog, not static snapshots.
The retention time of Changelog is controlled by the cluster parameter changelog.retention.ms, which defaults to 24 hours. You must ensure that the downstream consumption delay does not exceed this duration.
이전 주제: Flink Stream Write다음 주제: Flink Lookup Join

도움말 및 지원

문제 해결에 도움이 되었나요?

피드백