NiFi与Flink数据通道实战：Site-to-Site协议直连方案详解

1. 为什么要在 NiFi 和 Flink 之间搭一条数据通道

第一次接触这个组合的人往往会有点懵：NiFi 和 Flink 看起来都能做数据处理，为什么还要把它们连起来？我刚开始做实时数据管道的时候也纠结过这个问题，后来踩了几次坑才想明白——这两个东西压根不是竞争关系，而是上下游的互补关系。

NiFi 的强项在于数据采集、路由、格式转换和可视化编排。它天生就是干"数据搬运工"的活，几百种 Processor 覆盖了从文件、数据库、消息队列到 HTTP 接口的各种数据源，拖拖拽拽就能把一条采集链路搭起来，运维人员改个参数都不用写代码。但 NiFi 做有状态的复杂计算就很吃力，比如窗口聚合、CEP 事件匹配、大规模 join，这些用 NiFi 的 Record 处理能力硬扛，性能和可维护性都会崩。

Flink 恰好相反，它的强项是有状态的流式计算，窗口、状态管理、Exactly-Once 语义、事件时间处理都是看家本领。但 Flink 本身不擅长做数据采集和协议适配，你让它去连一堆五花八门的异构数据源，写起来又累又容易出错。

所以一个很自然的架构就出来了：NiFi 负责采集和预处理，Flink 负责计算和分析。问题就落在两者之间怎么传数据。常见方案有几种：走 Kafka 中转、走文件系统落地、走 HTTP 接口、还有就是本文要讲的Site-to-Site（站对站）。

Site-to-Site 是 NiFi 原生的一套协议，本来是为了多个 NiFi 实例之间互相传数据设计的。它的好处是自带流控、背压、可靠性保证，而且 Flink 官方提供了nifi-flink-connector，可以直接作为 Source 或 Sink 接入。相比走 Kafka 中转，少了一个中间件的运维成本；相比走文件落地，延迟低得多。这就是为什么很多团队会选择 S2S 直连。

这篇文章我会把整个链路的搭建过程拆开讲清楚：S2S 协议到底怎么工作、NiFi 端要开哪些配置、Flink 端连接器怎么用、参数怎么调、出问题怎么排查。适合已经对 NiFi 和 Flink 有基本了解、想打通两者数据通道的工程师参考，小白也能跟着一步步做下来。

2. Site-to-Site 协议到底是怎么工作的

2.1 从"两个 NiFi 互传"理解 S2S 的本质

要理解 S2S，最好的办法是先看它最初的设计场景：两个 NiFi 实例之间传数据。

假设 A 集群采集了数据，想传给 B 集群处理。A 上的一个 Processor 会把数据打包成FlowFile，然后通过 S2S 协议推送到 B。B 收到后，数据就进入了 B 的流程。整个过程里，A 是客户端（发起传输的一方），B 是服务端（接收的一方）。

S2S 协议的核心机制有这么几个：

• 传输端点（Transport Protocol）：底层可以用 RAW socket 或者 HTTP(S)。RAW 性能更好，HTTP 更容易穿透网络策略。
• 端口（Port）：服务端要显式开启一个"输入端口"或"输出端口"，客户端才能连。
• 流控与背压：客户端会先问服务端"你还能收多少"，服务端根据队列情况返回一个可接收的数量，客户端按这个数量发，避免把服务端压垮。
• 事务与确认：数据是分批传输的，每批传完要等服务端确认，确认后才算成功，失败会重试。

这套机制的好处是可靠性由协议本身保证，你不需要在应用层再写一套重试和确认逻辑。

2.2 Flink 连接器是怎么"伪装"成 NiFi 客户端的

Flink 的nifi-flink-connector本质上就是实现了一个 S2S 客户端。它内部封装了 S2S 的通信逻辑，对外暴露成标准的 Flink Source 和 Sink 接口。

作为Source时，Flink 扮演客户端角色，主动去 NiFi 的某个输出端口拉数据。这里有个容易搞反的点：NiFi 端要开的是 Output Port，不是 Input Port。因为数据是从 NiFi 流向 Flink，NiFi 是数据提供方。

作为Sink时，Flink 扮演客户端角色，把数据推到 NiFi 的输入端口。这时 NiFi 端要开的是 Input Port。

连接器的工作流程大致是：

1. Flink 启动时，通过配置的 URL 连接到 NiFi 实例，做一次握手，拿到端口信息。
2. 之后按批次拉取或推送数据，每批数据在 Flink 侧会被转换成FlowFile或从FlowFile解析出来。
3. 传输过程中连接器会维护事务状态，配合 Flink 的 Checkpoint 机制实现一致性。

2.3 为什么不用 Kafka 中转

很多人会问：既然有 Kafka，为什么还要用 S2S？我实测下来的对比是这样的：

如果你的数据量在每秒几万条以内，团队又不想多维护一套 Kafka，S2S 直连是很划算的选择。数据量再往上走，或者有多个下游消费者，那还是老老实实上 Kafka。

注意：S2S 直连的吞吐瓶颈主要在单个 NiFi 实例和单个连接上，如果 Flink 并行度很高，多个 subtask 同时连一个 NiFi 端口，NiFi 端可能会成为瓶颈。这时候要么给 NiFi 做集群，要么还是走 Kafka。

3. 环境准备与版本匹配的坑

3.1 版本兼容性是第一道坎

我踩过的第一个大坑就是版本不匹配。nifi-flink-connector这个连接器并不是 Flink 官方主仓库维护的，而是社区项目，版本更新没那么及时。你用的 Flink 版本和连接器版本必须严格对应，否则会出现各种诡异的类找不到、方法签名不匹配的问题。

我整理了一份实测可用的版本对应关系：

NiFi 版本建议 1.15 以上，因为早期版本的 S2S 协议在并发和流控上有一些已知问题。Flink 版本不要选太新的，连接器跟不上。

3.2 依赖引入的正确姿势

Maven 项目里引入连接器依赖，注意 scope 和打包方式：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-nifi</artifactId> <version>1.15.0</version> </dependency>

如果你用的是 Flink 1.15 以上，连接器可能不在 Maven 中央仓库，需要自己编译或者从社区仓库拉。编译的时候记得把 NiFi 的 S2S 相关依赖也打进去，否则运行时会报ClassNotFoundException。

打包成 fat jar 的时候，不要把 Flink 自身的依赖打进去，否则会和集群的类加载器冲突。用providedscope 标记 Flink 核心依赖。

3.3 网络与端口规划

S2S 通信需要网络可达，规划好端口：

• NiFi 的 S2S 默认走8080 端口（HTTP）或者自定义的 RAW socket 端口。
• 如果 NiFi 开了 HTTPS，端口通常是8443。
• Flink 集群的每个 TaskManager 都要能访问到 NiFi 的 S2S 端口，不是只有 JobManager 能访问就行。

实操心得：我建议先用telnet或curl从 Flink 的每个节点测一下 NiFi 端口通不通，别等作业提交了才发现网络不通。这一步能省掉大量排查时间。

4. NiFi 端配置：开端口、设权限、调流控

4.1 创建 Output Port 并配置

假设我们要让 Flink 从 NiFi 拉数据，NiFi 端要建一个 Output Port。

在 NiFi 画布上右键，选择Create Output Port，给它起个名字，比如to-flink。建好之后，把上游的 Processor 连到这个端口上。

关键配置项：

• Name：端口名，Flink 端配置时要对应上。
• Comments：写清楚用途，方便团队协作。
• Concurrent Tasks：并发任务数，默认 1。如果 Flink 并行度高，可以适当调大，但要注意 NiFi 的负载。
• Allow Remote Access：必须勾选，否则 Flink 连不上。这个选项在端口配置的 Settings 标签页里。

4.2 配置 S2S 的传输协议和端口

NiFi 的 S2S 配置在nifi.properties文件里，主要关注这几项：

# 是否启用 RAW socket 传输 nifi.remote.input.socket.enabled=true nifi.remote.input.socket.port=9999 # HTTP 传输配置 nifi.remote.input.http.enabled=true nifi.remote.input.http.port=8080 # 是否允许不安全的 HTTP nifi.remote.input.secure=false

如果 NiFi 开了 HTTPS，nifi.remote.input.secure要设为true，Flink 端也要配对应的信任库。

RAW socket 性能更好，但需要额外开端口；HTTP 更容易穿透网络策略。我一般优先用 HTTP，除非吞吐要求特别高。

4.3 权限与用户配置

S2S 连接需要认证。NiFi 默认用证书或者用户名密码。如果 NiFi 没开安全认证，那 Flink 端可以不配认证信息，但生产环境强烈建议开启。

开启认证后，需要在 NiFi 里创建一个用户，并给它访问 Output Port 的权限：

1. 进入Global Menu -> Users，创建用户。
2. 进入Policies，找到对应的 Output Port，把该用户加进去，赋予Read权限（作为 Source 时）或Write权限（作为 Sink 时）。

注意：权限配错是新手最常见的坑。现象是 Flink 报Access Denied或者连接超时，但网络明明是通的。遇到这种情况先检查 Policies 配置。

4.4 流控参数怎么调

NiFi 的 Output Port 有几个流控相关的参数，直接影响和 Flink 的配合：

• Back Pressure Object Threshold：队列里 FlowFile 数量超过这个值就触发背压，上游停止写入。默认 10000。
• Back Pressure Data Size Threshold：队列数据大小阈值，默认 1GB。
• FlowFile Expiration：FlowFile 过期时间，默认 0（不过期）。如果 Flink 消费慢，数据会堆积，建议设一个合理值，比如 7 天。

这些参数要和 Flink 的消费速率匹配。如果 Flink 消费慢，NiFi 队列会堆积，触发背压后上游采集也会变慢，这是正常的流控行为，不是 bug。

5. Flink 端开发：Source 和 Sink 的完整实现

5.1 用 NiFiSource 拉取数据

先看 Source 的写法。核心是构造NiFiSourceConfig，然后传给NiFiSource。

import org.apache.flink.streaming.connectors.nifi.NiFiSource; import org.apache.flink.streaming.connectors.nifi.NiFiSourceConfig; import org.apache.flink.streaming.connectors.nifi.StandardNiFiDataPacket; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class NiFiToFlinkJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(10000); // 10秒一次 checkpoint NiFiSourceConfig config = NiFiSourceConfig.builder() .url("http://nifi-host:8080/nifi") .portName("to-flink") .batchSize(100) // 每批拉取 100 个 FlowFile .build(); NiFiSource source = new NiFiSource(config); DataStream<String> stream = env .addSource(source) .map(new MapFunction<StandardNiFiDataPacket, String>() { @Override public String map(StandardNiFiDataPacket packet) { return new String(packet.getContent()); } }); stream.print(); env.execute("NiFi to Flink Job"); } }

几个关键点：

• url填 NiFi 的访问地址，注意结尾不要带斜杠。
• portName必须和 NiFi 端 Output Port 的名字完全一致，大小写敏感。
• batchSize控制每批拉取的数量，太小吞吐上不去，太大内存压力大。我一般从 100 开始调。
• enableCheckpointing必须开，否则 S2S 的事务确认机制没法配合，可能丢数据。

5.2 用 NiFiSink 推送数据

Sink 的写法类似，方向反过来：

import org.apache.flink.streaming.connectors.nifi.NiFiSink; import org.apache.flink.streaming.connectors.nifi.NiFiSinkConfig; import org.apache.flink.streaming.connectors.nifi.StandardNiFiDataPacket; NiFiSinkConfig sinkConfig = NiFiSinkConfig.builder() .url("http://nifi-host:8080/nifi") .portName("from-flink") .build(); NiFiSink<String> sink = new NiFiSink<>(sinkConfig, new NiFiSinkConverter<String>() { @Override public StandardNiFiDataPacket convert(String value) { return new StandardNiFiDataPacket( value.getBytes(), Collections.emptyMap() // 可以在这里加 FlowFile 属性 ); } }); stream.addSink(sink);

Sink 端 NiFi 要开的是Input Port，别搞反了。

5.3 数据格式转换的实战技巧

S2S 传的是FlowFile，内容是字节数组，属性是Map<String, String>。实际项目里数据格式转换是最容易出问题的地方。

我一般这么处理：

• NiFi 端：用ConvertRecord或UpdateAttribute把数据整理成统一的格式（比如 JSON），把元数据放到 FlowFile 属性里。
• Flink 端：在MapFunction里解析 JSON，把属性也读出来，转成 POJO。

public class FlowFileParser implements MapFunction<StandardNiFiDataPacket, Event> { private final ObjectMapper mapper = new ObjectMapper(); @Override public Event map(StandardNiFiDataPacket packet) throws Exception { Event event = mapper.readValue(packet.getContent(), Event.class); // 从属性里补充元数据 Map<String, String> attrs = packet.getAttributes(); event.setSource(attrs.getOrDefault("source", "unknown")); event.setTimestamp(Long.parseLong(attrs.getOrDefault("timestamp", "0"))); return event; } }

实操心得：FlowFile 的属性在传输过程中会保留，这是 S2S 相比纯字节流传输的一大优势。善用属性可以避免在数据体里塞元数据，减少解析开销。

5.4 并行度与连接数的关系

Flink 的并行度决定了有多少个 subtask 同时连 NiFi。每个 subtask 会独立建立 S2S 连接。

这里有个反直觉的点：并行度不是越高越好。NiFi 端的 Output Port 有并发限制，如果 Flink 并行度设成 10，但 NiFi 端口只允许 4 个并发连接，多出来的 subtask 会连不上或者排队。

我的经验是：

• NiFi 单实例，Flink 并行度设 2-4 比较合适。
• NiFi 集群，可以适当提高，但不要超过 NiFi 节点数乘以端口的并发任务数。
• 先小后大，观察 NiFi 的负载和队列情况再调。

6. 常见问题与排查技巧实录

6.1 连接类问题速查

6.2 数据丢失或重复

这是最让人头疼的问题。S2S 本身有事务保证，但配合 Flink 的 Checkpoint 时容易出问题。

数据丢失的常见原因：

• 没开 Checkpoint，S2S 的事务确认和 Flink 的状态不一致。
• Checkpoint 间隔太长，作业失败后回滚到很久之前的状态。
• NiFi 端 FlowFile 过期时间设太短，数据还没被消费就过期了。

数据重复的常见原因：

• Checkpoint 失败后重试，S2S 已经确认的数据被重新拉取。
• 没有实现幂等处理。

解决办法：

1. 开启 Checkpoint，间隔设 10-30 秒。
2. 下游处理实现幂等，比如用唯一 ID 去重。
3. NiFi 端 FlowFile 过期时间设长一点，至少覆盖 Checkpoint 间隔的几倍。

6.3 性能调优的几个关键参数

6.4 我踩过的三个真实坑

坑一：NiFi 开了 HTTPS，Flink 没配信任库。现象是连接一直超时，日志里只有一句模糊的 SSL 错误。后来在 Flink 的启动参数里加了-Djavax.net.ssl.trustStore才解决。

坑二：并行度设太高，NiFi 端口并发不够。现象是部分 subtask 一直处于INITIALIZING状态，作业起不来。把并行度从 8 降到 4 就好了。

坑三：FlowFile 属性里有中文，传输后乱码。S2S 传输属性时默认用 ISO-8859-1 编码，中文会乱。解决办法是在 NiFi 端把属性值做 URL 编码，Flink 端解码。

提示：排查 S2S 问题时，NiFi 的nifi-app.log和 Flink 的 TaskManager 日志要一起看。很多问题在单边日志里看不出来，两边对照才能定位。

7. 生产环境部署的几点经验

7.1 高可用怎么考虑

NiFi 单点故障会导致 Flink 作业拉不到数据。生产环境建议 NiFi 做集群，Flink 端配置多个 URL：

NiFiSourceConfig config = NiFiSourceConfig.builder() .url("http://nifi1:8080/nifi,http://nifi2:8080/nifi,http://nifi3:8080/nifi") .portName("to-flink") .build();

连接器会自动做负载均衡和故障转移。注意 NiFi 集群的节点要配置成同一个集群，端口名要一致。

7.2 监控指标看什么

NiFi 端重点看：

• Output Port 的队列长度和背压状态。
• S2S 的连接数和传输速率。
• 系统负载和 JVM 内存。

Flink 端重点看：

• Source 的numRecordsOut和numBytesOut。
• Checkpoint 的成功率和耗时。
• 反压指标（backPressure）。

两边指标对不上，比如 NiFi 显示发了 100 万条，Flink 只收到 90 万，那中间肯定有问题，要查日志。

7.3 数据一致性怎么保证

要做到端到端 Exactly-Once，需要：

1. Flink 开启 Checkpoint。
2. NiFi 端的 FlowFile 在确认后才删除（S2S 默认行为）。
3. 下游 Sink 支持事务或幂等。

如果下游是数据库，用两阶段提交或者幂等写入。如果下游是文件，用临时文件加原子重命名。

实际项目里，做到 At-Least-Once 加幂等处理，往往比追求 Exactly-Once 更实际。因为 Exactly-Once 的性能开销不小，而且实现复杂。

7.4 容量规划参考

根据我的经验，单 NiFi 实例配单 Flink 作业（并行度 4），S2S 直连的吞吐大概在：

• 小消息（1KB 左右）：每秒 2-5 万条。
• 中等消息（10KB）：每秒 5000-1 万条。
• 大消息（100KB）：每秒 500-1000 条。

超过这个量级，要么加 NiFi 节点，要么考虑 Kafka 中转。别硬扛，S2S 不是为超高吞吐设计的。

8. 一个完整的端到端示例

8.1 场景描述

假设我们要做一个实时日志分析：NiFi 采集服务器日志，做初步清洗后通过 S2S 推给 Flink，Flink 做关键词统计和异常检测。

8.2 NiFi 流程设计

1. ListenHTTP或TailFile采集日志。
2. ReplaceText清洗格式。
3. UpdateAttribute添加来源、时间戳属性。
4. 连到 Output Portlog-to-flink。

8.3 Flink 作业实现

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(15000); env.setParallelism(4); NiFiSourceConfig config = NiFiSourceConfig.builder() .url("http://nifi-cluster:8080/nifi") .portName("log-to-flink") .batchSize(200) .build(); DataStream<LogEvent> logs = env .addSource(new NiFiSource(config)) .map(new LogParser()); // 关键词统计 DataStream<Tuple2<String, Integer>> keywordCount = logs .flatMap(new KeywordExtractor()) .keyBy(value -> value.f0) .window(TumblingProcessingTimeWindows.of(Time.minutes(1))) .sum(1); keywordCount.print(); env.execute("Log Analysis Job");

8.4 验证与压测

上线前一定要做压测。我的做法是：

1. 用 NiFi 的GenerateFlowFile造数据，模拟真实流量。
2. 逐步加大流量，观察 NiFi 队列和 Flink 反压。
3. 记录吞吐拐点，确定生产环境的容量上限。
4. 模拟 NiFi 节点宕机，验证故障转移。

压测时特别注意 Checkpoint 的耗时，如果 Checkpoint 时间超过间隔，作业会一直处于 Checkpoint 中，吞吐会掉得厉害。

9. 写在最后的一点个人体会

这套 S2S 直连方案我在几个项目里都用过，最大的感受是：它适合快速打通链路，但不适合当长期大规模方案。中小规模、团队人手紧、不想多维护中间件的时候，S2S 直连非常香，一两天就能跑起来。但数据量一旦上去，或者业务对可靠性要求极高，还是老老实实上 Kafka 或者 Pulsar 更稳妥。

另外提醒一句，nifi-flink-connector这个社区项目的活跃度一般，遇到问题查资料会比较费劲。建议把关键版本的源码拉下来读一读，尤其是 S2S 事务处理那部分，理解了原理排查问题会快很多。我当初就是靠读源码才搞明白 Checkpoint 和 S2S 确认之间的时序关系，文档里根本没写清楚。

如果你也在用这套方案，欢迎交流踩坑经验。特别是高并发场景下的调优，我这边数据量还没到那个量级，经验有限。