pipeline.max-parallelism: 2048containerized.taskmanager.env.HADOOP_USER_NAME: hadoopcontainerized.master.env.HADOOP_USER_NAME: hadooptaskmanager.memory.managed.fraction: 0.1execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_SUCCESSexecution.checkpointing.tolerable-failed-checkpoints: 5flink.hadoop.fs.cosn.trsf.fs.AbstractFileSystem.ofs.impl: com.qcloud.chdfs.fs.CHDFSDelegateFSAdapterflink.hadoop.fs.cosn.trsf.fs.ofs.impl: com.qcloud.chdfs.fs.CHDFSHadoopFileSystemAdapterflink.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: trueflink.hadoop.fs.AbstractFileSystem.cosn.impl: org.apache.hadoop.fs.CosNflink.hadoop.fs.cosn.impl: org.apache.hadoop.fs.CosFileSystemflink.hadoop.fs.cosn.bucket.region: <YourRegion>flink.hadoop.fs.cosn.userinfo.appid: <YourAppId>flink.hadoop.fs.ofs.impl: com.qcloud.chdfs.fs.CHDFSHadoopFileSystemAdapterflink.hadoop.fs.ofs.tmp.cache.dir: /tmp/chdfs/flink.hadoop.fs.ofs.upload.flush.flag: trueflink.hadoop.fs.ofs.user.appid: <YourAppId>flink.hadoop.fs.ofs.bucket.region: <YourRegion>
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');
参数 | 必填 | 说明 | 默认值 |
connector | 是 | 固定为 setats | (无，必须填写) |
manager-url | 是 | Setats Manager 地址 | (无，必须填写) |
warehouse | 是 | Setats Warehouse 路径 | (无，必须填写) |
catalog-type | 是 | Catalog 类型，常用 Hive 或 Hadoop | (无，必须填写) |
catalog-name | 是 | Catalog 名称 | (无，必须填写) |
catalog-database | 是 | 数据库名称 | default |
catalog-table | 建议填写 | 目标表名，建议与实际表名保持一致 | (无) |
location | 建议填写 | 表的物理存储路径，使用 COS / HDFS 时建议显式指定 | (无) |
uri | Hive Catalog 必填 | Hive Metastore 地址 | (无，必须填写) |
bucket | 否 | 目标表的桶数量，建议结合数据规模和并发写入能力配置 | -1 |
bucket-key | 否 | 分桶键，用于决定数据写入时落到哪个 Bucket。建议选择分布较均匀、相对稳定的字段，通常可使用主键字段或主键子集 | (无) |
merge-engine | 否 | 合并策略： DEDUPLICATE：合并取最后一条 PARTIAL_UPDATE：部分列更新，取每个字段最后一个不为空的值 | DEDUPLICATE |
setats.sink.flush-interval | 否 | Sink 刷新间隔，例如 5s | 5s |
-- 创建测试数据源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');-- 创建 Setats Sink 表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');-- 执行写入INSERT INTO `demo_setats_table1`SELECT`id`,DATE_FORMAT(NOW(), 'yyyyMMdd') AS `dt`,`name`,`address`FROM `test_source_datagen`;
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');
文档反馈