PyFlink与NiFi数据流直接集成方案咨询：无需Kafka中转

方案1：NiFi HTTP处理器 + PyFlink自定义HTTP Source #

• NiFi端：配置SendHTTP处理器，将数据以POST请求形式发送到PyFlink暴露的HTTP端点，可根据JSON/CSV等数据格式设置请求头与请求体。
• PyFlink端：自定义HTTP Source，通过启动轻量HTTP服务监听指定端口，接收NiFi请求并解析数据送入流处理管道。示例代码思路：

  from pyflink.datastream import StreamExecutionEnvironment
  from pyflink.datastream.functions import SourceFunction
  import threading
  from flask import Flask, request
  
  app = Flask(__name__)
  data_queue = []
  
  @app.route('/ingest', methods=['POST'])
  def ingest_data():
      data = request.get_data().decode('utf-8')
      data_queue.append(data)
      return 'OK', 200
  
  class HTTPSource(SourceFunction):
      def run(self, ctx):
          while True:
              if data_queue:
                  ctx.collect(data_queue.pop(0))
  
      def cancel(self):
          pass
  
  if __name__ == '__main__':
      env = StreamExecutionEnvironment.get_execution_environment()
      threading.Thread(target=app.run, kwargs={'host': '0.0.0.0', 'port': 8080}).start()
      ds = env.add_source(HTTPSource())
      ds.print()
      env.execute("NiFi-PyFlink HTTP Ingest")
  

• 生产环境需注意：请求幂等性、重试机制、HTTP服务高可用性。

方案2：NiFi Socket处理器 + PyFlink内置Socket Source #

• NiFi端：使用PutTCP/PutUDP处理器，将数据发送到PyFlink监听的TCP/UDP端口。
• PyFlink端：直接用内置socket_text_stream作为数据源，无需额外开发：

  env = StreamExecutionEnvironment.get_execution_environment()
  ds = env.socket_text_stream('localhost', 9999)
  ds.print()
  env.execute("NiFi-PyFlink Socket Ingest")
  

• 局限性：Socket传输缺乏可靠性保障（如丢包、无重传），适合对一致性要求较低的场景，生产环境需配合NiFi的重试与数据持久化配置。

方案3：基于NiFi Site-to-Site协议对接 #

NiFi的Site-to-Site（S2S）协议支持高效点对点传输，可通过以下方式集成：

• 自定义PyFlink Source，借助jpype/py4j调用NiFi官方Java版S2S客户端API，从NiFi输出端口拉取数据。
• 核心步骤：
  1. 在NiFi中配置输出端口并开启Site-to-Site；
  2. 在PyFlink中通过Java桥接初始化S2S客户端，建立连接后拉取数据流。
• 优势：自带流量控制、数据可靠性保障，适合生产环境。

方案4：NiFi File处理器 + PyFlink File Source #

• NiFi端：用PutFile处理器将数据写入共享存储（HDFS/本地文件系统）指定目录，配置文件滚动策略（如按大小、时间间隔）。
• PyFlink端：使用read_text_file监听目录，读取新增文件：

  env = StreamExecutionEnvironment.get_execution_environment()
  ds = env.read_text_file("/path/to/nifi/output", watchType="PATTERN")
  ds.print()
  env.execute("NiFi-PyFlink File Ingest")
  

• 注意：确保PyFlink集群可访问共享存储，配置合理的文件滚动与清理策略，避免目录堆积过多文件。

内容的提问来源于stack exchange，提问作者Umut Bilgen