tencent cloud

Stream Compute Service

DocumentaçãoStream Compute ServiceStream Lake Engine SetatsJob Development GuideDeveloping Flink JobsFlink Stream Write

Flink Stream Write

Download
Modo Foco
Tamanho da Fonte
Última atualização: 2026-07-03 15:10:13

Overview

Setats supports continuously writing real-time data through Flink by using the Setats connector to write Upsert data into Setats primary key tables. After data is written, it triggers Eager compaction and generates a Changelog. This enables the system to concurrently meet requirements for scenarios such as second-level visibility, subsequent CDC consumption, and OLAP queries.
Note:
1. This example demonstrates the following job, which uses a Hive catalog and a COS acceleration bucket.
2. If you use a Hadoop catalog, you must modify the 'catalog-type'='hadoop'.
3. When using Oceanus, you do not need to create tables in advance because the Flink Setats Sink automatically creates them.

Prerequisites

You have created a Setats cluster and completed the Warehouse configuration.
You have obtained the manager-url.
You have prepared the Flink runtime environment, for example, SCS.
You have confirmed that the underlying storage associated with the Warehouse, the Hive Metastore, network access, and permission configurations are all properly connected.
If you use a COS acceleration bucket as the Warehouse, we recommend that you additionally verify the following:
The target directory, for example, cosn://<bucket>/warehouse, has been created in advance within the acceleration bucket.
You have configured access permissions for the acceleration bucket to the subnet where Oceanus resides.
The Flink cluster has the network capability to access Hive Metastore, COS / CHDFS.

Oceanus Advanced Parameter Examples

If your job runs on Oceanus and the Warehouse uses a COS acceleration bucket, you can refer to the following advanced parameter template:
pipeline.max-parallelism: 2048
containerized.taskmanager.env.HADOOP_USER_NAME: hadoop
containerized.master.env.HADOOP_USER_NAME: hadoop
taskmanager.memory.managed.fraction: 0.1
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_SUCCESS
execution.checkpointing.tolerable-failed-checkpoints: 5
flink.hadoop.fs.cosn.trsf.fs.AbstractFileSystem.ofs.impl: com.qcloud.chdfs.fs.CHDFSDelegateFSAdapter
flink.hadoop.fs.cosn.trsf.fs.ofs.impl: com.qcloud.chdfs.fs.CHDFSHadoopFileSystemAdapter
flink.hadoop.fs.cosn.trsf.fs.ofs.tmp.cache.dir: /tmp/chdfs/
flink.hadoop.fs.cosn.trsf.fs.ofs.user.appid: <YourAppId>
flink.hadoop.fs.cosn.trsf.fs.ofs.bucket.region: <YourRegion>
flink.hadoop.fs.cosn.trsf.fs.ofs.upload.flush.flag: true
flink.hadoop.fs.AbstractFileSystem.cosn.impl: org.apache.hadoop.fs.CosN
flink.hadoop.fs.cosn.impl: org.apache.hadoop.fs.CosFileSystem
flink.hadoop.fs.cosn.bucket.region: <YourRegion>
flink.hadoop.fs.cosn.userinfo.appid: <YourAppId>
flink.hadoop.fs.ofs.impl: com.qcloud.chdfs.fs.CHDFSHadoopFileSystemAdapter
flink.hadoop.fs.ofs.tmp.cache.dir: /tmp/chdfs/
flink.hadoop.fs.ofs.upload.flush.flag: true
flink.hadoop.fs.ofs.user.appid: <YourAppId>
flink.hadoop.fs.ofs.bucket.region: <YourRegion>
Note:
Replace parameters such as `AppId`, `Region`, and `HADOOP_USER_NAME` according to your actual environment. The available parameter items may vary slightly across different cluster environments. Please refer to the verification results from your actual runtime environment.

Table Creation Syntax

Creating a Setats Target Table

The following example uses Hive Catalog as an example:
CREATE TABLE `setats_sink` (
  `id` STRING,
  `dt` STRING,
  `name` STRING,
  `address` STRING,
  PRIMARY KEY (`id`, `dt`) NOT ENFORCED
) PARTITIONED BY (`dt`) WITH (
  'connector' = 'setats',
  'manager-url' = '<Manager Url>',
  'catalog-type' = 'hive',
  'catalog-name' = 'setats',
  'catalog-database' = 'testdb',
  'catalog-table' = 'setats_sink',
  'warehouse' = 'cosn://<bucket>/warehouse',
  'location' = 'cosn://<bucket>/warehouse/testdb/setats_sink',
  'uri' = 'thrift://<metastore-host-1>:7004,thrift://<metastore-host-2>:7004',
  'bucket' = '10',
  'bucket-key' = 'id',
  'merge-engine' = 'PARTIAL_UPDATE',
  'setats.sink.flush-interval' = '5s'
);

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
Recommended
Target table name. It is recommended to keep it consistent with the actual table name.
(None)
location
Recommended
Physical storage path of the table. It is recommended to explicitly specify it when COS / HDFS is used.
(None)
uri
Required for Hive Catalog.
Hive Metastore address
(Not provided, must be specified)
bucket
No
Number of buckets for the target table. It is recommended to configure this based on data volume and concurrent write capability.
-1
bucket-key
No
Bucket key, which determines which bucket the data falls into during writes. It is recommended to select a field with relatively uniform distribution and stability. Primary key fields or subsets of primary keys are commonly used.
(None)
merge-engine
No
Merge policy:
DEDUPLICATE: merges and retains the last record.
PARTIAL_UPDATE: partially updates columns, taking the last non-null value for each field.
DEDUPLICATE
setats.sink.flush-interval
No
Sink flush interval, for example, 5s.
5s

Write Examples

-- Create a test data source

CREATE TABLE `test_source_datagen` (
  `id` STRING,
  `name` STRING,
  `address` STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.id.start' = '0',
  'fields.id.end' = '100000',
  'fields.id.kind' = 'sequence',
  'fields.name.length' = '5',
  'fields.address.length' = '10'
);
-- Create a Setats Sink table

CREATE TABLE `demo_setats_table1` (
  `id` STRING,
  `dt` STRING,
  `name` STRING,
  `address` STRING,
  PRIMARY KEY (`id`, `dt`) NOT ENFORCED
) PARTITIONED BY (`dt`) WITH (
  'connector' = 'setats',
  'manager-url' = '<Manager Url>',
  'catalog-type' = 'hive',
  'catalog-name' = 'setats',
  'catalog-database' = 'testdb',
  'catalog-table' = 'demo_setats_table1',
  'warehouse' = 'cosn://<bucket>/warehouse',
  'location' = 'cosn://<bucket>/warehouse/testdb/demo_setats_table1',
  'uri' = 'thrift://<metastore-host-1>:7004,thrift://<metastore-host-2>:7004',
  'bucket' = '10',
  'merge-engine' = 'PARTIAL_UPDATE',
  'setats.sink.flush-interval' = '5s'
);
-- Perform writing

INSERT INTO `demo_setats_table1`
SELECT
  `id`,
  DATE_FORMAT(NOW(), 'yyyyMMdd') AS `dt`,
  `name`,
  `address`
FROM `test_source_datagen`;

Partial Column Updates

The Setats primary key table supports partial column updates. In a Flink write job, you can declare only the columns to be updated. Columns not present in the write record retain their original values. For business scenarios such as orders, profiles, and user Tags that involve "incrementally supplementing columns by primary key", it is recommended to use this feature in conjunction with merge-engine = 'PARTIAL_UPDATE'.
Example:
CREATE TABLE `setats_partial_sink` (
  `id` STRING,
  `dt` STRING,
  `name` STRING,
  PRIMARY KEY (`id`, `dt`) NOT ENFORCED
) PARTITIONED BY (`dt`) WITH (
  'connector' = 'setats',
  'manager-url' = '<Manager Url>',
  'catalog-type' = 'hive',
  'catalog-name' = 'setats',
  'catalog-database' = 'testdb',
  'catalog-table' = 'demo_setats_table1',
  'warehouse' = 'cosn://<bucket>/warehouse',
  'uri' = 'thrift://<metastore-host-1>:7004,thrift://<metastore-host-2>:7004',
  'merge-engine' = 'PARTIAL_UPDATE'
);

Publishing and Running

When SQL jobs are published in managed environments such as Oceanus, it is recommended to perform checks in the following order:
1. Check whether the dependency JAR has been uploaded and associated with the current space.
2. Check whether the Hive configuration JAR has been referenced along with the job.
3. Check whether Warehouse, location, uri, and manager-url have been replaced with actual values.
4. Check whether the relevant advanced parameters have been correctly configured for the acceleration bucket scenario.

Notes

Design partition fields such as dt based on business query patterns to avoid excessive partitioning incurring additional management overhead.
The number of buckets should be comprehensively evaluated based on write concurrency, data volume, and the number of Workers.
Parameters such as manager-url, warehouse, and location must be consistent with the cluster configuration.
Anterior: Configuration ManagementPróximo: Flink CDC Stream Reading

Ajuda e Suporte

Esta página foi útil?

comentários