如何使用Python版Flink（PyFlink）从HDFS读取数据

PyFlink本身没有直接暴露HDFS InputFormat的Python API，但可以通过包装Java InputFormat的方式实现，这是官方兼容跨语言场景的常用方案，具体步骤如下：

1. 确认依赖 #

确保Flink集群及PyFlink环境包含flink-hadoop-compatibility依赖包，版本需与你的Flink版本（如1.15）匹配。本地运行时可将依赖包放入PyFlink的classpath，集群提交时通过--jar参数指定。

2. 编写Python代码包装Java InputFormat #

通过PyFlink的Java网关调用Hadoop的InputFormat类，将其包装为Python可使用的数据源：

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import InputFormatSourceFunction
from pyflink.java_gateway import get_gateway

# 初始化执行环境
env = StreamExecutionEnvironment.get_execution_environment()

# 获取Java网关，调用Hadoop相关类
gateway = get_gateway()
HadoopConf = gateway.jvm.org.apache.hadoop.conf.Configuration
TextInputFormat = gateway.jvm.org.apache.hadoop.mapreduce.lib.input.TextInputFormat
HdfsPath = gateway.jvm.org.apache.hadoop.fs.Path

# 配置HDFS路径与Hadoop参数
hdfs_data_path = "hdfs://your-nn-host:9000/path/to/target/data"
conf = HadoopConf()
# 若需Kerberos认证等，添加对应配置，例如：conf.set("hadoop.security.authentication", "kerberos")

# 创建TextInputFormat实例并绑定配置
input_format = TextInputFormat(HdfsPath(hdfs_data_path))
input_format.setConf(conf)

# 包装为PyFlink数据源并添加到环境
hdfs_source = InputFormatSourceFunction(input_format)
data_stream = env.add_source(hdfs_source)

# 示例：打印读取到的数据
data_stream.print()

# 执行任务
env.execute("PyFlink HDFS Read Task")

3. 处理结构化数据 #

如果HDFS存储的是Parquet、ORC等结构化格式，只需替换对应的Java InputFormat类（如org.apache.parquet.hadoop.ParquetInputFormat），后续通过map算子将读取的字节流解析为Python对象即可。

4. 关键注意事项 #

• 确保Hadoop配置文件（core-site.xml、hdfs-site.xml）可被Flink集群访问，或在代码中通过HadoopConf显式设置NameNode地址等核心参数。
• PyFlink版本必须与Flink集群版本完全一致，避免兼容性问题。
• YARN集群运行时，可通过flink run的--yarnship参数上传Hadoop依赖包，确保任务节点能加载相关类。

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