1. 先说版本现状：Flink 2.2 目前还没有可用的 ES Connector 依赖
如果你在看 Flink 2.2 的官方文档，会看到一个非常关键的提示：
- DataStream 的 Elasticsearch 6.x/7.x connector：Flink 2.2 暂无可用 connector(Apache Nightlies)
- Table/SQL 的 Elasticsearch connector：Flink 2.2 也暂无可用 connector(Apache Nightlies)
但在 Flink 1.20 这类稳定版本，ES 连接器是可用的，并且官方文档给出了明确的 Maven 坐标（例如3.1.0-1.20）。 (Apache Nightlies)
你写博客时可以直接点明：
想在 Flink 2.2 用 ES sink，要么等待 2.2 对应连接器发布，要么短期选用已发布连接器的稳定版本（例如 1.20 / 2.0 对应的独立 connector 版本），避免“文档有、依赖没有”的尴尬。 (Apache Nightlies)
2. 依赖怎么选：按 ES 版本选 6 或 7（示例基于 Flink 1.20）
Flink 1.20 的文档给出了 DataStream connector 的依赖示例：
- ES 6.x：
flink-connector-elasticsearch6 - ES 7.x：
flink-connector-elasticsearch7
版本示例：3.1.0-1.20(Apache Nightlies)
你可以在博客里贴这个（以 ES7 为例）：
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-elasticsearch7</artifactId><version>3.1.0-1.20</version></dependency>3. 最小可运行：写入 ES 的第一条数据（IndexRequest）
Flink 的 ES Sink 走的是Elasticsearch6SinkBuilder / Elasticsearch7SinkBuilder，核心是setEmitter：你把每条流数据转换成 ES 的请求，然后indexer.add(request)。 (Apache Nightlies)
ES 7 示例（官方风格）
input.sinkTo(newElasticsearch7SinkBuilder<String>().setBulkFlushMaxActions(1)// 每条都 flush，演示用；生产别这么配.setHosts(newHttpHost("127.0.0.1",9200,"http")).setEmitter((element,context,indexer)->indexer.add(createIndexRequest(element))).build());privatestaticIndexRequestcreateIndexRequest(Stringelement){Map<String,Object>json=newHashMap<>();json.put("data",element);returnRequests.indexRequest().index("my-index").id(element).source(json);}ES6 和 ES7 的最大差异之一是：ES6 示例里还有.type("my-type")，ES7 不需要。 (Apache Nightlies)
4. 内部机制：BulkProcessor 才是“吞吐的灵魂”
Flink 的 ES Sink 在每个并行子任务内部都维护一个 BulkProcessor：
- 先把 action 请求缓存起来
- 再按条件批量 flush 到 ES
- 并且一次只会执行一个 bulk（不会并发 flush）(Apache Nightlies)
这意味着两件事：
- 你调吞吐，本质上是在调 BulkProcessor 的 flush 策略
- 你的并行度越高，总体写入吞吐通常越高（前提：ES 也扛得住）
5. 可容错语义：Checkpoint 打开后是 At-least-once
官方文档明确：启用 checkpoint 后，ES Sink 能保证at-least-once，做法是 checkpoint 时等待 BulkProcessor 中 pending 的请求全部被 ES ack。 (Apache Nightlies)
启用方式很简单：
StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(5000);还有一个容易踩坑的点：
checkpoint 默认是不开的，但 connector 的默认投递语义是 AT_LEAST_ONCE，这会导致数据先缓存在 BulkProcessor 里，默认攒到 1000 个 action 才 flush（或者等自动 flush 条件触发）。 (Apache Nightlies)
如果你发现“数据进 ES 很慢”，第一件事别怀疑人生，先看你是不是没开 checkpoint、并且 flush 条件太大。
6. “准 Exactly-once”的工程套路：deterministic id + upsert
文档给了一个非常实用的结论：
在 AT_LEAST_ONCE 的前提下，如果你用UpdateRequest + deterministic id + upsert，可以把最终效果做到“看起来像 exactly-once”。 (Apache Nightlies)
工程化翻译一下就是：
同一条业务记录无论被重试写几次，最终落在 ES 里都是同一个_id，写入是幂等覆盖，不会出现重复文档。
你在博客里可以强调两条最佳实践：
_id一定要可复现（比如业务主键、traceId、聚合窗口 key 等）- 更新用 upsert（不存在就插入，存在就更新），把重试成本变成幂等写
7. 失败重试与背压：Backoff 好用，但会拉长 Checkpoint
ES 写入失败的原因常见两类：
- 临时资源不足（比如节点队列满、线程池饱和）
- 请求本身有问题（比如文档字段类型不匹配、非法数据）
Flink ES Sink 支持配置 backoff 策略，让“资源不足”类错误重试，例如指数退避： (Apache Nightlies)
.setBulkFlushBackoffStrategy(FlushBackoffType.EXPONENTIAL,5,1000)但要注意文档的警告：
失败请求被重新加入 BulkProcessor，会让 checkpoint 变长，因为 checkpoint 也要等待这些 re-add 的请求 flush 完成。 (Apache Nightlies)
实战建议（很管用）：
- ES 偶发抖动：开 backoff，重试次数别太离谱
- ES 长期扛不住：别靠重试硬顶，应该降写入速率或扩容 ES（不然 checkpoint 会被拖到崩）
8. BulkProcessor 调参指南：三件套 + 重试策略
官方给了 BulkProcessor 的关键可调项： (Apache Nightlies)
setBulkFlushMaxActions(n)：攒多少条 action flushsetBulkFlushMaxSizeMb(mb)：攒到多大 flushsetBulkFlushInterval(ms)：不管攒多少，到了时间就 flushsetBulkFlushBackoffStrategy(type, retries, delay)：临时错误的重试策略（常量/指数退避）
一套比较“稳”的生产配置思路（给你写博客用）：
- 低延迟优先：
maxActions小一点 +interval短一点（例如 200~500ms），吞吐会下降但延迟更稳 - 高吞吐优先：
maxActions大一点 +maxSizeMb控住（避免单 bulk 太大），延迟会上升但 ES 压力更均匀 - ES 容量紧张：
maxInFlight（如果你的版本/实现有）要控住，并且 backoff 打开
9. PyFlink 也能用：记得加 JAR
Flink 文档也给了 PyFlink 的依赖说明：需要额外把flink-connector-elasticsearch6/7的 JAR 带上，否则运行时找不到类。 (Apache Nightlies)
10. 打包上线：Uber-Jar 或放到 Flink lib
最后是上线必做项：连接器默认不在 Flink 二进制发行包里，所以你要么做 uber-jar，把依赖打进一个可执行 jar，要么把 connector jar 放进 Flink 的lib/目录让集群全局可见。 (Apache Nightlies)
你写 CSDN 时，这段建议直接写到“部署注意事项”，基本能挡住 80% 的“本地能跑、集群 ClassNotFound”的问题。
11. 顺手补一段：Table/SQL 连接器的能力点（但 2.2 暂无依赖）
即使是 Table/SQL Connector，Flink 也支持两种模式：
- DDL 有主键：upsert 模式，可消费 UPDATE/DELETE
- DDL 无主键：append 模式，只能 INSERT (Apache Nightlies)
并且 SQL connector 还有 failure-handler 策略（fail/ignore/retry-rejected/自定义类）以及动态 index 等能力点。 (Apache Nightlies)
但同样要强调：Flink 2.2 文档目前标注为“暂无可用 connector 依赖”，这块更适合写成“能力预告 + 迁移规划”。 (Apache Nightlies)
)
)
