1、先把连接器看成“能力块”

Flink SQL 里常见连接器可以按能力拆开理解：

• Source（Scan）：读全量/读增量
• Lookup Source：维表查（Temporal Join）
• Sink：写外部系统
• Append vs Upsert：是否能吃 UPDATE/DELETE（是否需要 PRIMARY KEY）

你贴的这些连接器里，有几个最常用的“组合拳”：

• JDBC：既能做 Scan，也能做 Lookup 维表，还能做 Sink；有主键就 Upsert（更适合容错重放的幂等写）
• Elasticsearch / OpenSearch：典型 Sink；有主键走 Upsert，主键会拼成 document_id；还支持动态 index（常用于按天分索引）
• FileSystem：既能批读也能流式 watch 目录；写入时最关键是滚动策略、小文件治理、分区提交（_SUCCESS / metastore）
• HBase：强 upsert 思维（rowkey 就是主键），列族用 ROW 映射，维表 join 很常见（可配缓存）
• Hive：两条线
  1）HiveCatalog 把 HMS 当元数据中枢（表一次建好，跨会话复用）
  2）Flink 作为引擎读写 Hive 表，支持批+流；流读可监控新分区/新文件，分区提交也能做 metastore + success-file (Apache Nightlies)
• OpenAI：把 LLM 推理封装成 Flink SQL 的 MODEL + ML_PREDICT，可做文本分类/抽取/Embedding 等；示例仍用 chat completions / embeddings 端点 (Apache Nightlies)

2、“最短闭环”三件套：DataGen / Print / BlackHole

2.1 DataGen：可控造数，专治“没数据难调 SQL”

用 DataGen 建一个流式表，控制速率、行数、字段分布（随机/序列），就能稳定复现问题。

CREATE TABLE gen_orders (
  order_id BIGINT,
  user_id  BIGINT,
  amount   DECIMAL(10,2),
  ts       TIMESTAMP(3),
  proctime AS PROCTIME()
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '5000',
  'fields.order_id.kind' = 'sequence',
  'fields.order_id.start' = '1',
  'fields.order_id.end'   = '100000000',
  'fields.user_id.min' = '1',
  'fields.user_id.max' = '200000',
  'fields.amount.min' = '1',
  'fields.amount.max' = '999'
);

2.2 Print：先验正确性（看 row_kind + 数据形态）

Print sink 会把每条记录打印到 Task 日志，格式里会带 RowKind（+I / -U / +U / -D），非常适合确认“你写的 SQL 产生的是 append 还是 changelog”。

CREATE TABLE sink_print (
  user_id BIGINT,
  cnt     BIGINT,
  sum_amt DECIMAL(20,2)
) WITH (
  'connector'='print',
  'print-identifier'='CHECK'
);

验证 SQL（比如聚合）：

INSERT INTO sink_print
SELECT
  user_id,
  COUNT(*) AS cnt,
  SUM(amount) AS sum_amt
FROM gen_orders
GROUP BY user_id;

你要看的点很明确：

• 是否出现 -U/+U（更新流）
• 数值是否符合预期（比如 sum 是否越来越大）
• 是否有 NULL/脏值导致的异常分支

2.3 BlackHole：压测吞吐（不让 Sink 干扰你）

BlackHole 直接吞数据，类似 Linux 的 /dev/null。用它压测，能把瓶颈更聚焦地落在：算子本身、状态、序列化、网络 shuffle、checkpoint 上。

CREATE TABLE sink_bh (
  user_id BIGINT,
  cnt     BIGINT,
  sum_amt DECIMAL(20,2)
) WITH (
  'connector'='blackhole'
);

INSERT INTO sink_bh
SELECT
  user_id,
  COUNT(*) AS cnt,
  SUM(amount) AS sum_amt
FROM gen_orders
GROUP BY user_id;

这就是“同一段 SQL，Print 看对不对，BlackHole 测快不快”的最短闭环。

3、一套通用压测模板（join / agg / topn / UDF 都能套）

你把真实要压测的 SQL 填到这里即可：

-- 1) 源：datagen / kafka / filesystem 都行，这里先用 datagen
-- 2) （可选）维表：先用 datagen 或 VALUES 模拟，再替换 JDBC/HBase/Hive lookup
-- 3) 目标：先 print 看对，再 blackhole 压测

-- 正确性验证
INSERT INTO sink_print
SELECT /* 你的 SQL 核心输出 */
FROM ...
;

-- 性能压测（完全同一份逻辑）
INSERT INTO sink_bh
SELECT /* 同上 */
FROM ...
;

你会得到两类“非常干净”的结论：

• Print：语义是否正确、changelog 是否符合预期
• BlackHole：极限吞吐在哪、是否被状态/网络/反压拖死

4、从闭环迁移到真实系统：几个最常见落地组合

4.1 JDBC 维表 Lookup：最常见的“事实流 + 维表补全”

CREATE TABLE dim_user (
  id BIGINT,
  name STRING,
  status BOOLEAN,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'='jdbc',
  'url'='jdbc:mysql://localhost:3306/app',
  'table-name'='user'
);

CREATE TABLE sink_print_enriched (
  order_id BIGINT,
  user_id  BIGINT,
  user_name STRING,
  amount DECIMAL(10,2)
) WITH ('connector'='print');

INSERT INTO sink_print_enriched
SELECT
  o.order_id,
  o.user_id,
  u.name,
  o.amount
FROM gen_orders o
LEFT JOIN dim_user FOR SYSTEM_TIME AS OF o.proctime AS u
ON o.user_id = u.id;

上线前仍然建议：先把 dim_user 换成 VALUES 或 datagen 维表，把 SQL 跑通；再切回 JDBC。

4.2 Hive：把 HMS 当“表的注册中心”，把 Hive 表当“批流统一仓”

Hive 这块你贴的内容非常关键：

• 流读可以监控新分区/新文件
• 维表 join 可以用“最新分区”作为维表版本（适合日更维表） (Apache Nightlies)

示例（思路版）：

-- 建 catalog（依赖 hive-site.xml）
CREATE CATALOG myhive WITH (
  'type'='hive',
  'hive-conf-dir'='/opt/hive-conf'
);

USE CATALOG myhive;

-- Hive 维表（每天一个分区版本）
-- streaming-source.partition.include='latest'：只取最新分区作为维表版本
-- streaming-source.monitor-interval：多久刷新一次

读写 Hive 表时，最容易踩的坑是：

• 监控间隔太小导致 metastore 压力大（尤其 join 场景会频繁 refresh）
• 分区提交策略没配好，导致下游读到未完整数据（需要 delay + watermark/partition-time） (Apache Nightlies)

4.3 OpenAI：把“文本理解/分类/Embedding”变成 SQL 算子

Flink 的 OpenAI Model Function 允许你用 SQL 声明一个 MODEL，然后 ML_PREDICT 调用推理。(Apache Nightlies)

注意：OpenAI 的 Chat Completions/Embeddings 端点仍可用，但官方文档也提示“新项目优先使用 Responses API”。(OpenAI 平台)
另外，模型名称也在演进，建议以官方 Models 列表为准。(OpenAI 平台)

情感分类（你贴的示例风格）：

CREATE MODEL ai_analyze_sentiment
INPUT (`input` STRING)
OUTPUT (`content` STRING)
WITH (
  'provider'='openai',
  'endpoint'='https://api.openai.com/v1/chat/completions',
  'api-key'='<YOUR_KEY>',
  'model'='gpt-4o-mini',
  'system-prompt'='Classify into [positive, negative, neutral, mixed]. Output only the label.',
  'temperature'='0',
  'n'='1'
);

CREATE TEMPORARY TABLE print_sink(
  id BIGINT,
  movie_name STRING,
  predict_label STRING,
  actual_label STRING
) WITH ('connector'='print');

INSERT INTO print_sink
SELECT id, movie_name, content AS predict_label, actual_label
FROM ML_PREDICT(
  TABLE movie_comment,
  MODEL ai_analyze_sentiment,
  DESCRIPTOR(user_comment)
);

Embedding：

CREATE MODEL ai_embed
INPUT (`input` STRING)
OUTPUT (`vec` ARRAY<FLOAT>)
WITH (
  'provider'='openai',
  'endpoint'='https://api.openai.com/v1/embeddings',
  'api-key'='<YOUR_KEY>',
  'model'='text-embedding-3-small'
);

生产建议（非常重要）：

• 先用 Print 验证输出 schema、失败策略、空值处理，再切真实 sink
• 控制成本：n=1、temperature=0、设置 max-tokens、必要时对输入做截断
• 错误处理：RETRY/IGNORE/FAILOVER 选型要和业务容错一致（尤其是外部 API 限流、超时） (OpenAI 平台)

5、Checklist：从“能跑”到“能稳”

1）主键策略

• 需要幂等写的 sink（JDBC/ES/OpenSearch）尽量定义 PRIMARY KEY，让 upsert 生效
• 没主键就只能 append，容错重放时更容易重复/冲突

2）反压与瓶颈定位（BlackHole 压测时最清晰）

• 看算子 backpressure、busy time、records in/out
• 如果开了状态：关注 state size、checkpoint time、RocksDB compaction（如用 RocksDB）

3）Hive 场景

• streaming-source.monitor-interval 不要太激进
• 分区提交用 partition-time + delay 更“准”，但要求 watermark/时间抽取配齐 (Apache Nightlies)

4）函数与性能

• HiveModule 能把 Hive 内置函数带进 Flink；聚合函数在某些场景可开启 native agg 获取更好的性能（sum/count/avg/min/max 等） (Apache Nightlies)

5）AI 推理

• 选模型、端点、速率限制、重试策略都要“可控”；把 API Key 放到安全配置体系里（别写死在脚本/仓库）