Flink OpenSearch SQL Connector Append/Upsert、动态索引、Exactly-Once 与性能调参

1. Connector 能做什么

OpenSearch SQL Connector 是一个 Sink（写入端）：

支持 Batch Sink
支持 Streaming Sink
支持 Streaming 的 Append & Upsert Mode

关键规则很简单：

DDL 定义了PRIMARY KEY：Sink 走Upsert，能承接带 UPDATE/DELETE 的 changelog 流
DDL 没定义主键：只能Append（仅 INSERT），无法正确消费 UPDATE/DELETE

2. 版本与依赖：Flink 2.2 的注意点

文档明确写了：Flink 2.2 目前还没有（yet）可用的 connector 发行物，并且 OpenSearch connector 不在 Flink 二进制发行包里。

工程上意味着两件事：

你跑集群任务时，必须把 connector 依赖打进 uber-jar 或放进 Flink 的lib/目录（让集群全局可见）
如果你使用的是 Flink 2.2，要提前确认你实际环境里能拿到对应版本的 connector jar（很多团队会固定到一个已发布可用的 Flink/connector 组合）

3. 5 分钟上手：创建 OpenSearch Sink 表

最小可用 DDL（写入到静态 index）：

CREATETABLEmyUserTable(user_id STRING,user_name STRING,uvBIGINT,pvBIGINT,PRIMARYKEY(user_id)NOTENFORCED)WITH('connector'='opensearch','hosts'='http://localhost:9200','index'='users');

然后你就可以把任意上游表的结果写进去：

INSERTINTOmyUserTableSELECTuser_id,user_name,uv,pvFROMsome_agg_table;

4. Append vs Upsert：一切从主键开始

这类“写搜索引擎”的 connector，最容易踩的坑就是：上游 SQL 产生了更新流，但下游表没主键，导致写入语义对不上。

怎么判断上游是不是会产生 UPDATE/DELETE？

聚合（尤其是持续聚合）、去重、TopN、Join 等都很容易产出 changelog
只要你的结果不是纯 INSERT-only，就强烈建议为 sink 表定义主键，让 connector 进入 Upsert 模式

5. 文档 ID 生成规则：主键会变成`_id`

OpenSearch connector 会用主键来计算 document id：

会把所有主键字段按 DDL 顺序拼接成一个字符串
拼接分隔符由document-id.key-delimiter控制，默认_
document id 有限制：最大 512 bytes、不能有空白字符
一些类型不适合做主键（例如 BYTES、ROW、ARRAY、MAP 等），因为字符串表示不稳定或不直观

示例：复合主键 + 自定义分隔符

CREATETABLEuserMetrics(tenant_id STRING,user_id STRING,uvBIGINT,pvBIGINT,PRIMARYKEY(tenant_id,user_id)NOTENFORCED)WITH('connector'='opensearch','hosts'='http://localhost:9200','index'='users','document-id.key-delimiter'='$');

最终 id 会长这样：tenantA$user123

实战建议：

主键字段尽量短、稳定、无空白、无易变格式
如果主键很长（尤其拼接后可能超过 512 bytes），要在业务侧设计更短的 key（例如 hash、短码、数值 id）

6. Dynamic Index：按字段/按日期路由写入

index支持静态与动态两种写法：

静态：'users'
动态：'index-{field_name}'
动态日期格式：'users-{log_ts|yyyy-MM-dd}'
甚至可以用now()：'users-{now()|yyyy-MM-dd}'

6.1 基于字段时间分索引（推荐）

典型日志场景：

CREATETABLEods_logs_opensearch(log_id STRING,log_tsTIMESTAMP(3),levelSTRING,msg STRING,PRIMARYKEY(log_id)NOTENFORCED)WITH('connector'='opensearch','hosts'='http://localhost:9200','index'='logs-{log_ts|yyyy-MM-dd}');