1. 为什么要在 NiFi 和 Flink 之间搭一条数据通道数据管道做久了你会发现一个很现实的问题采集层和计算层往往不是同一套技术栈。采集侧用 Apache NiFi 的团队非常多因为它的可视化编排、背压控制、数据溯源能力在异构数据源接入场景下确实省心而计算侧一旦涉及有状态流处理、事件时间窗口、Exactly-Once 语义Flink 又是绕不开的选择。两套系统各自都很强但把它们拼在一起的时候很多人第一反应是中间加个 Kafka 不就行了。加 Kafka 当然能跑通但它引入了一个额外的中间件运维成本、延迟、数据一致性边界都多了一层。NiFi 从 1.0 版本开始就内置了Site-to-Site站对站简称 S2S协议这个协议的设计初衷就是让多个 NiFi 实例之间高效、可靠地传输数据。而 Flink 生态里有一个相对低调但非常实用的连接器——nifi 连接器它让 Flink 能够以 S2S 客户端或服务端的身份直接和 NiFi 对话。也就是说NiFi 和 Flink 可以跳过 Kafka直接建立一条点对点的数据通道。这条通道适合谁如果你手上有 NiFi 集群在跑数据采集下游想用 Flink 做实时 ETL、实时特征计算或者流式聚合又不想为了传个数据再维护一套消息队列那 S2S 方案值得认真评估。它尤其适合数据量中等偏大、对延迟敏感、且希望减少组件数量的场景。但要注意S2S 不是万能的它有明确的适用边界后面我会详细拆解。这篇文章我会从协议原理讲到实际配置包括 NiFi 端怎么开 S2S、Flink 端连接器怎么选参数、端口和权限怎么配、批量大小和超时怎么调以及我在实际联调中踩过的几个坑。内容基于常见实践整理具体参数需要你根据自己的集群规模做调整。2. Site-to-Site 协议到底在传什么2.1 S2S 不是简单的 HTTP 文件传输很多人第一次听到 Site-to-Site会以为它就是 NiFi 之间用 HTTP 传文件。这个理解只对了一半。S2S 确实底层走 HTTP(S)但它上面封装了一套完整的事务性数据传输协议。核心概念有三个FlowFile、Transaction、Port。FlowFile 是 NiFi 里数据的基本单位你可以把它理解成内容 属性的包裹内容是实际字节流属性是元数据键值对。S2S 传输的不是裸文件而是带着完整属性的 FlowFile。Transaction 保证的是传输的原子性——一批 FlowFile 要么全部被接收方确认要么全部回滚重传不会出现传了一半的中间状态。Port 则是接收端的入口分为Input Port接收数据和Output Port发送数据。Flink 连接器在这套体系里扮演的是客户端角色。它通过 S2S 协议向 NiFi 的 Output Port 发起请求NiFi 把 FlowFile 推送给 FlinkFlink 确认收到后 NiFi 才删除或转移这些数据。这个确认后删除的机制是 S2S 可靠性的关键也是它区别于普通 HTTP 拉取的地方。2.2 两种角色Flink 当客户端还是当服务端Flink 的 nifi 连接器支持两种模式这个选择直接影响你的网络配置和部署方式。第一种NiFi 作为 S2S 服务端Flink 作为客户端。这是最常见的模式。NiFi 端配置一个 Output Port 并启用 S2SFlink 端配置 NiFi 的地址、端口名主动去拉数据。网络方向是 Flink 所在节点能访问 NiFi 的 S2S 端口默认 8080 或你自定义的端口。这种模式的好处是 NiFi 不需要知道 Flink 在哪里Flink 可以动态扩缩容NiFi 侧配置一次就行。第二种Flink 作为 S2S 服务端NiFi 作为客户端。这种模式用得少但在某些网络拓扑下有用——比如 NiFi 集群在 DMZ 区Flink 在内网只允许 NiFi 主动往外连。这时候 Flink 连接器需要暴露一个 S2S 服务端口NiFi 端配置一个 Remote Process Group 指向 Flink。配置复杂度明显更高而且 Flink 作业重启时端口会变需要额外做服务发现。提示除非有明确的网络方向限制否则优先选第一种模式。第二种模式在 Flink 作业生命周期管理上会带来不少麻烦我后面会讲具体坑在哪。2.3 数据流的方向和背压传导S2S 协议本身是拉模式。Flink 客户端主动向 NiFi 请求数据NiFi 根据请求里带的参数比如一次要多少条、超时多久返回一批 FlowFile。这个拉模式天然带背压Flink 处理不过来的时候请求频率下降NiFi 侧的 Output Port 队列就会堆积NiFi 的背压机制会进一步向上游传导最终让采集端降速。这个特性在实时管道里非常重要。如果用 Kafka 做中间层背压是通过消费者 lag 间接体现的你需要额外监控和告警。而 S2S 的背压是端到端的NiFi 的 UI 上能直接看到队列堆积情况排查起来直观得多。但要注意NiFi 的 Output Port 队列如果堆满且配置了背压阈值上游 Processor 会被暂停这时候如果上游是采集关键数据可能会丢数据或者阻塞采集。所以队列容量和背压阈值的配置需要根据你的数据重要性和下游处理能力来权衡。3. NiFi 端的 S2S 配置从端口到权限3.1 开启 S2S 并配置接收端口NiFi 默认的 S2S 配置在nifi.properties文件里关键参数有这么几个# S2S 传输协议端口默认和 UI 共用 8080 nifi.remote.input.socket.port # 是否启用 RAW 协议Socket 直连性能更高 nifi.remote.input.securetrue # HTTP 传输时的端口 nifi.remote.input.http.enabledtrue nifi.remote.input.http.port8080这里有个容易混淆的点S2S 支持两种传输协议——HTTP和RAW。HTTP 走的是和 UI 相同的端口配置简单穿透防火墙容易RAW 走独立的 Socket 端口省去了 HTTP 头开销吞吐量更高但需要额外开放端口。如果你的数据量不大每秒几千条以内HTTP 完全够用如果每秒几万条以上建议开 RAW。配置 RAW 端口时nifi.remote.input.socket.port填一个未被占用的端口比如 9999。然后在 NiFi 的 UI 上进入Controller Settings → Site to Site确认 Input Port 和 Output Port 的配置。注意S2S 的端口是在 Process Group 级别创建的你需要在画布上添加一个Output Port节点给它起个名字比如to-flink这个名字后面 Flink 端要用到。3.2 用户认证和授权别跳过这一步S2S 连接需要认证。NiFi 支持多种认证方式和 Flink 连接器配合时最常用的是用户名密码和客户端证书两种。用户名密码方式配置简单在 NiFi 的login-identity-providers.xml里确保有可用的身份提供者然后给 Flink 使用的账号分配权限。权限分两层全局权限能否访问 S2S和端口权限能否读写特定端口。在 UI 的Users页面创建用户后需要在Policies里给这个用户授予 retrieve site-to-site details 和对应 Output Port 的 receive data 权限。客户端证书方式更安全适合生产环境。你需要生成客户端证书导入 NiFi 的 truststore然后在 NiFi 里配置证书身份映射。Flink 端配置 keystore 和 truststore 路径。这种方式配置繁琐但避免了密码在配置文件中明文存储的问题。注意我见过好几次联调失败是因为只配了全局 S2S 权限忘了给具体 Output Port 授权。NiFi 的权限模型是细粒度的全局权限和组件权限是分开的排查时先看 NiFi 的nifi-user.log里面会明确记录权限拒绝的原因。3.3 Output Port 的队列和背压设置Output Port 本身有几个关键配置项直接影响 Flink 拉取时的行为配置项默认值建议调整说明Back Pressure Object Threshold10000根据下游消费能力调整队列中 FlowFile 数量超过此值时触发背压Back Pressure Data Size Threshold1 GB根据单条数据大小调整队列数据总量超过此值时触发背压FlowFile Expiration0永不过期按业务设置如 24 hours防止下游长期不消费导致数据无限堆积Concurrent Tasks1可适当提高提高并发拉取能力但受限于下游背压阈值的设置逻辑是这样的假设 Flink 每秒能处理 5000 条NiFi 上游每秒产生 6000 条那么队列会以每秒 1000 条的速度增长。如果阈值设成 10000大约 10 秒后触发背压上游降速。这个缓冲时间够不够取决于你的上游采集能否承受短暂的暂停。如果上游是数据库 CDC 采集暂停可能导致 binlog 堆积如果上游是日志文件 tail暂停只是延迟读取问题不大。FlowFile Expiration 这个参数特别容易被忽略。默认 0 表示永不过期如果 Flink 作业挂了三天没人管NiFi 队列里堆了三天的数据恢复后 Flink 会一次性拉取大量积压数据可能直接把 Flink 作业打爆。设置一个合理的过期时间比如 24 小时能让系统在异常时自动丢弃过老的数据保护下游。4. Flink 连接器的参数拆解与选型逻辑4.1 连接器依赖和版本匹配Flink 的 nifi 连接器不是内置的需要额外引入依赖。Maven 坐标大致是这样的dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-nifi/artifactId version${flink.version}/version /dependency版本匹配是个大坑。连接器版本必须和 Flink 运行时版本严格对应比如 Flink 1.17 的作业只能用 1.17 系列的连接器。跨大版本基本必挂跨小版本有时候能跑但会有奇怪的序列化问题。我建议在pom.xml里用${flink.version}统一管理避免手动指定版本号导致不一致。另外连接器底层依赖 NiFi 的nifi-site-to-site-client库这个库的版本和 NiFi 服务端版本也有兼容性要求。一般来说客户端库版本不低于服务端版本就能正常工作但反过来客户端比服务端老很多可能会遇到协议不兼容。生产环境建议客户端库版本和服务端 NiFi 版本保持一致或略高。4.2 核心参数逐个说清楚Flink 的NiFiSource构造参数不算多但每个都影响行为NiFiSourceConfig config NiFiSourceConfig.builder() .url(https://nifi-host:8080/nifi) .portName(to-flink) .username(flink-user) .password(******) .requestBatchSize(1000) .requestDelay(0) .build();url填 NiFi 的访问地址注意如果 NiFi 开了 HTTPS这里要用https://而且证书要被 Flink 的 JVM truststore 信任。自签名证书需要额外导入否则会报 SSL 握手失败。portName就是 NiFi 画布上那个 Output Port 的名字必须完全一致大小写敏感。我遇到过因为名字里多了个空格导致连接建立失败的情况排查了半天。requestBatchSize是一次请求拉取的 FlowFile 数量。这个值不是越大越好。设太大单次请求的数据量可能超过 Flink 单个 checkpoint 能处理的范围导致 checkpoint 超时设太小请求频率过高S2S 的事务开销占比上升吞吐下降。经验值在 500 到 5000 之间具体要看单条 FlowFile 的大小。如果单条平均 1KB1000 条就是 1MB比较合适如果单条 1MB那 1000 条就是 1GB明显太大了。requestDelay是两次请求之间的间隔毫秒数。默认 0 表示不等待拉完一批立刻拉下一批。这个参数在限流场景下有用——比如你想控制 Flink 的消费速率可以设一个正值。但要注意这个 delay 是在 Flink 端控制的NiFi 端并不知道所以它不能替代 NiFi 的背压机制。4.3 并行度和 Source 算子的关系Flink 的 NiFi Source 默认并行度是 1。这意味着即使你的 Flink 集群有几十个 TaskManager这个 Source 也只有一个 subtask 在拉数据。对于数据量大的场景这会是明显的瓶颈。能不能提高并行度可以但有个前提NiFi 端的 Output Port 要支持并发拉取。S2S 协议本身支持多个客户端同时连接同一个 Output PortNiFi 会把队列里的 FlowFile 分发给不同的客户端。所以你可以把 Flink Source 的并行度设成 2 或 4每个 subtask 独立建立 S2S 连接各自拉取一部分数据。但这里有个权衡并行度提高后数据在 Flink 内部的分布会变化。如果下游算子依赖 key 的哈希分布Source 并行度变化会导致数据重新分布可能影响状态大小和性能。另外每个 subtask 都会维护自己的 S2S 连接和事务状态并行度太高会增加 NiFi 端的连接管理开销。我一般建议从 2 开始试观察 NiFi 的 S2S 连接数和 Flink 的吞吐再决定是否继续加。5. 联调过程中最容易卡住的几个环节5.1 连接建立失败从日志定位问题联调第一步就是连接建立这一步失败的概率最高。Flink 端报的错通常比较笼统比如Unable to connect to NiFi具体原因要看 NiFi 端的日志。排查顺序我一般是这样的网络连通性从 Flink 所在节点telnet nifi-host 8080确认端口通。如果不通先查防火墙和安全组。NiFi S2S 是否启用检查nifi.properties里nifi.remote.input.http.enabledtrue改完要重启 NiFi。认证是否通过看 NiFi 的nifi-user.log如果看到Authentication failed说明用户名密码或证书有问题。权限是否足够如果认证过了但报Access denied看nifi-user.log里具体缺哪个权限去 Policies 里补。端口名是否正确确认 Flink 配置的portName和 NiFi 画布上的 Output Port 名字完全一致。这个顺序是从底层到上层能避免在权限问题上浪费时间——如果网络都不通查权限毫无意义。5.2 数据拉到了但内容不对属性丢失和编码问题连接通了数据也拉到了但发现 FlowFile 的属性没了或者内容乱码。这两个问题很常见。属性丢失通常是因为 Flink 连接器默认只读取 FlowFile 的内容属性需要显式配置才会带上。在NiFiSourceConfig里有一个readAttributes之类的开关不同版本叫法可能不同开启后属性会作为元数据附加到 Flink 的Record或Row上。但要注意属性是键值对Flink 的流式 API 里没有原生的属性概念你需要把它们映射成数据字段或者侧输出。编码问题多半是 NiFi 端 FlowFile 的内容编码和 Flink 端解析编码不一致。NiFi 的 FlowFile 内容是字节流没有编码信息Flink 端需要自己指定。如果 NiFi 上游写的是 UTF-8Flink 端用默认的 ISO-8859-1 解析中文就会乱码。解决办法是在 Flink 的 DeserializationSchema 里明确指定StandardCharsets.UTF_8。5.3 作业重启后的状态恢复问题Flink 作业从 checkpoint 恢复时NiFi Source 的状态恢复是个需要特别注意的点。S2S 协议本身不提供从某个偏移量继续消费的能力它只有确认已收到的机制。这意味着如果 Flink 作业在拉取了一批数据但还没完成 checkpoint 时挂了恢复后这批数据可能会被重新拉取至少一次语义或者丢失如果 NiFi 已经删除了数据。要实现精确一次需要依赖 Flink 的 checkpoint 机制和 NiFi 端的数据保留策略配合。具体来说NiFi 的 Output Port 在数据被 S2S 客户端确认接收后会删除 FlowFile如果 Flink 在确认后、checkpoint 完成前挂了这批数据就丢了。要避免这个问题NiFi 端可以配置成确认后不立即删除而是转移到另一个队列保留一段时间但这会增加 NiFi 的存储压力。实际生产中大多数场景能接受至少一次语义配合下游的幂等处理来保证最终一致。如果业务要求严格精确一次建议在 NiFi 和 Flink 之间加一层可重放的消息队列或者使用支持事务的 S2S 扩展方案。这个取舍需要在架构设计阶段就想清楚不要等到上线后才发现语义不满足。6. 性能调优从吞吐量到资源占用6.1 批量大小和请求频率的平衡前面提到requestBatchSize和requestDelay这两个参数是性能调优的主要抓手。我做过一组简单的压测在单条 FlowFile 平均 2KB、NiFi 和 Flink 在同一局域网的条件下不同配置的吞吐表现大致如下requestBatchSizerequestDelay (ms)吞吐量 (条/秒)平均延迟 (ms)1000约 8000155000约 220002510000约 350004020000约 420007050000约 45000150100010约 1800060可以看到批量从 100 提到 1000吞吐翻了四倍多延迟只增加了不到 30ms。但从 2000 提到 5000吞吐增长放缓延迟却翻倍。这是因为单次请求的数据量增大后网络传输时间和 Flink 内部的反序列化时间都线性增长而 S2S 的事务开销被摊薄后边际收益递减。我的建议是延迟敏感型场景选 500-1000吞吐优先型场景选 2000-3000超过 5000 基本没必要。requestDelay除非有明确的限流需求否则保持 0。6.2 NiFi 端的并发和 JVM 调优Flink 端调优只是一半NiFi 端的配置同样影响整体吞吐。Output Port 的Concurrent Tasks默认是 1意味着只有一个线程在处理 S2S 请求。如果 Flink 端并行度是 4NiFi 端只有 1 个线程就会成为瓶颈。把 Concurrent Tasks 调到和 Flink 并行度相当的水平能明显提升吞吐。NiFi 的 JVM 堆大小也需要关注。S2S 传输过程中FlowFile 的内容会在内存中缓冲如果堆太小频繁 GC 会拖慢传输。一般来说NiFi 堆大小建议不低于 4GB数据量大时 8-16GB 更稳妥。但也不是越大越好堆太大导致 GC 停顿时间变长反而影响实时性。可以用 G1 GC 并设置MaxGCPauseMillis来控制停顿。另外NiFi 的nifi.content.repository.archive.max.usage.percentage参数控制内容仓库的归档阈值S2S 传输频繁时内容仓库的 IO 压力不小建议把内容仓库放在 SSD 上和 FlowFile 仓库、Provenance 仓库分开磁盘。6.3 监控指标看什么、在哪看S2S 链路的监控分两端。NiFi 端主要看 Output Port 的队列大小、背压状态、S2S 连接数。这些在 NiFi UI 的 Summary 页面和 Bulletin 里都能看到。如果队列持续增长且触发背压说明 Flink 消费能力不足如果连接数频繁波动可能是网络不稳定或 Flink 作业在重启。Flink 端主要看 Source 算子的numRecordsOut、numBytesOut指标以及 checkpoint 的持续时间和大小。如果 checkpoint 时间突然变长可能是 Source 拉取的数据量增大导致状态膨胀。Flink 的 Web UI 和 Metrics Reporter 都能采集这些指标建议接入 Prometheus Grafana 做长期监控和告警。还有一个容易被忽略的指标是 NiFi 的Provenance数据。S2S 传输的每个 FlowFile 都会生成 Provenance 事件数据量大时 Provenance 仓库会快速增长影响 NiFi 性能。可以适当降低 Provenance 的采样率或者缩短保留时间在可追溯性和性能之间取平衡。7. 这套方案适合什么场景不适合什么场景7.1 适合的场景特征S2S 直连方案在以下几种情况下优势明显组件精简优先。团队规模不大运维力量有限不想为了数据传输再维护一套 Kafka 集群。NiFi 和 Flink 都是已有组件S2S 是 NiFi 内置能力Flink 连接器也是官方维护的没有引入新的中间件。数据量中等且稳定。每秒几万条到几十万条NiFi 和 Flink 都在同一机房或低延迟网络内。这个量级下 S2S 的吞吐完全够用而且延迟比经过 Kafka 中转更低。需要端到端背压。NiFi 的背压能直接传导到采集端不需要额外的监控和限流逻辑。对于采集速率波动大的场景这个特性很实用。数据溯源要求高。NiFi 的 Provenance 能记录每条数据的完整流转路径S2S 传输也在溯源范围内。对于需要审计数据来源的场景这比 Kafka 的方案更直观。7.2 不适合的场景和替代方案反过来以下场景建议慎重考虑 S2S 直连需要精确一次语义且不能容忍重复。前面分析过S2S 原生不支持偏移量管理精确一次需要额外机制。如果业务对重复零容忍Kafka 的事务生产者 Flink 的 Kafka Source 是更成熟的选择。跨机房或跨地域传输。S2S 对网络延迟比较敏感跨地域的高延迟会导致事务超时和重传。这种情况下在两地各部署一套 NiFi中间用 Kafka 或对象存储做桥接更稳妥。数据需要被多个下游消费。S2S 是点对点传输一个 Output Port 的数据被 Flink 拉走后NiFi 端就删除了。如果还有别的下游也要这份数据就需要在 NiFi 里做复制分发或者改用 Kafka 做发布订阅。Flink 作业频繁重启或动态扩缩容。每次重启都要重新建立 S2S 连接和认证如果作业本身不稳定频繁的连接建立会加重 NiFi 负担。Kafka 的消费者组机制在这种场景下更从容。提示如果你的场景介于适合和不适合之间可以先做一个最小验证——用少量数据跑通链路观察延迟、吞吐和稳定性再决定是否上生产。不要一上来就全量切换。8. 几个我踩过的坑和对应的解法8.1 NiFi 重启后 Flink 作业卡死NiFi 集群做滚动重启时Flink 作业的 S2S 连接会断开。连接器有重试机制但如果 NiFi 重启时间较长超过了连接器的重试超时Flink 作业会进入失败状态。更麻烦的是有些版本里连接断开后 Flink 作业不会自动恢复而是一直卡在RUNNING状态但不再拉取数据。解法是在 Flink 端配置合理的重试策略包括重试次数和退避时间。同时给 Flink 作业开启Restart Strategy比如固定延迟重启让作业在连接失败后能自动重启并重新建立连接。NiFi 端做滚动重启时尽量逐个节点操作保证 S2S 端口始终有可用节点。8.2 大批量数据导致 checkpoint 超时有一次上游突然产生了一波历史数据回填NiFi 队列里瞬间堆了几百万条 FlowFile。Flink 的 Source 并行度是 1requestBatchSize是 5000每次拉取的数据在 Flink 内部缓冲checkpoint 时需要把这些缓冲数据也纳入状态快照导致 checkpoint 时间从正常的几百毫秒飙升到几十秒最终超时失败。解法有两个方向一是提高 Source 并行度把拉取压力分散到多个 subtask二是降低requestBatchSize减少单次拉取的数据量让 checkpoint 的状态更小。另外可以在 Flink 的 checkpoint 配置里适当增大超时时间但这只是缓解不是根治。根治还是要控制 NiFi 队列的堆积量设置合理的 FlowFile 过期时间和背压阈值。8.3 属性中的特殊字符导致解析异常NiFi 的 FlowFile 属性里如果包含换行符、制表符或者 JSON 特殊字符Flink 端解析时可能会出错。我遇到过一次上游在属性里写了一个包含换行符的 SQL 语句Flink 端用默认的文本格式解析时直接把一行拆成了多行导致数据错位。解法是在 Flink 端使用更健壮的解析方式比如 JSON 格式而不是分隔符格式或者在 NiFi 端用ReplaceText等 Processor 对属性做清洗把特殊字符转义或替换掉。如果属性值确实需要保留原始格式可以在 Flink 端用自定义的 DeserializationSchema 做针对性处理。8.4 证书过期导致的静默失败生产环境用客户端证书认证时证书是有有效期的。证书过期后S2S 连接会失败但 Flink 端的报错可能只是SSL handshake failed不会直接告诉你证书过期了。更隐蔽的是如果 NiFi 端配置了多个身份提供者证书认证失败后可能会 fallback 到其他方式导致行为不一致。解法是建立证书到期提醒机制在证书过期前至少一个月更新。同时在 Flink 端的日志里开启 SSL 调试-Djavax.net.debugssl能看到详细的握手过程快速定位是证书问题还是其他 SSL 配置问题。NiFi 端的nifi-user.log也会记录认证失败的详细信息两边对照看能省不少时间。9. 写在最后的一点个人体会这套 S2S 直连方案我在几个项目里用过整体感受是它适合简单直接的场景不适合复杂多变的场景。如果你的数据管道拓扑稳定、数据量可预期、团队对 NiFi 和 Flink 都比较熟S2S 能省掉一个中间件链路更短排查问题也更直接。但如果你的场景涉及多下游、跨地域、频繁变更那 Kafka 这样的消息中间件虽然多了一层但换来的灵活性和生态成熟度是值得的。技术选型没有绝对的对错关键是清楚每个方案的边界在哪里。S2S 的边界就是点对点、同地域、中等数据量、可接受至少一次语义。在这个边界内它是个很优雅的方案出了边界硬撑只会给自己找麻烦。最后分享一个小技巧联调阶段可以先用 NiFi 的GenerateFlowFileProcessor 造一批测试数据属性里带上时间戳和序列号Flink 端把序列号打印出来这样能快速验证数据是否完整、是否有重复、延迟大概多少。等链路稳定了再接入真实数据源能避免很多到底是链路问题还是数据源问题的扯皮。