Flink读Parquet与Avro避坑：FileSource批流一体与PyFlink依赖冲突

前一阵在某数据平台项目里处理一堆日志和画像数据，被“Flink 读 Parquet + Avro + FileSource 批流一体 + PyFlink 依赖”这条链路折腾得够呛。文档里每个组件单拎出来都挺正常，一旦拼在一起，各种隐藏的坑就全冒出来了。这块内容拆开看都不算新，但组合起来确实有门槛，尤其是 RowData 和 Avro 的类型映射、FileSource 在批流两种模式下的行为差异、PyFlink 加依赖时的 ClassLoader 冲突，这三块几乎每个环节都值得单独写一篇。

这篇文章就把我当时踩过的坑、验证过的方案、最后沉淀下来的一整套配置和代码思路完整梳理一遍。不管你是刚接触 Flink 批流一体，还是已经在用 Flink + Parquet 做离线数仓，只要涉及 Avro/Parquet 格式解析、FileSource 数据源接入、或者 PyFlink 混用 Java/Scala 依赖，这篇文章应该都能帮你省下不少“试错时间”。

1. 为什么这条链路这么容易卡壳——先认清三个角色各自的立场

在动手写代码之前，我建议你先停下来想清楚一个问题：Flink 读 Parquet，本质上是“谁在跟谁打交道”？

这条链路里至少有三个角色：外部存储里的数据文件（Parquet/Avro 格式）、Flink 的类型系统（核心是 RowData）、以及你用来声明格式的 API（Table/SQL 的 DESCRIBE、DataStream 的 DecodingFormat）。很多人之所以一上来就懵，是因为把这三个角色的职责搞混了。

Parquet 本身不关心你的 Java 对象是 RowData 还是 GenericRecord。Parquet 只是一种列式存储格式，文件里保存的是经过编码的列数据，外加一个 schema。真正决定“读出来是什么对象”的，是 Flink 的格式插件（Format）在反序列化之后往哪个类型上转换。也就是说，同样是读 Parquet 文件，你可以在 DataStream API 里得到 RowData，也可以在 Table API 里得到内部的行结构，这两者的 API 使用方式完全不同，但底层都是同一套 parquet 列解码逻辑。

Avro 和 Parquet 的关系更微妙。Avro 算是一种“带 schema 的序列化框架”，它本身既可以作为存储格式（Avro 文件），也可以作为其他存储格式（比如 Parquet）的“逻辑类型来源”。市面上常见的做法是：用 Avro schema 来定义数据的结构（字段名、类型、嵌套关系），然后让 Parquet 按这个 schema 来落盘。这就带来一个常见问题：你在 Hadoop 生态里用 Spark 写出的 Parquet 文件，schema 是 Avro 风格的；但是 Flink 的 parquet 格式插件的 schema 其实是从 Table schema 或文件自带的 schema 转换来的，两边的“字段类型方言”不一致，于是读出来就出现类型不匹配、精度丢失、甚至直接解析失败。

FileSource 是 Flink 1.12+ 主推的新数据源架构。它跟旧的 StreamingFileSource / ContinuousFileMonitoringFunction 不是一个物种。FileSource 把“文件列表的发现（Listing）”和“文件内容的读取（Reader）”拆成了两件事，借助 SplitEnumerator / SourceReader 的架构实现。批模式下它会把所有文件一次性列出来分片执行；流模式下它会周期性扫描目录或对象存储新增的文件。这个差异直接决定了你写代码时要不要给“流式读文件”做额外处理。

我用一个表格把这几个角色理清一下：

实际开发中，我见过很多团队把这三层混为一谈，结果就是：换了 Parquet 文件版本，读不出来了；加了嵌套字段，反序列化抛异常了；批任务切到流任务，新文件永远读不到了。下面按我的实际排查顺序，从格式层、数据源层、API 层逐个展开说。

2. Parquet 接入 FileSource：批流一体到底是自动的还是配置出来的

2.1 FileSource 的基本形态和两种模式的行为差异

Flink 官方文档里老说 FileSource 支持批流一体，但“支持”这个词其实是有条件的。FileSource 本身有两种模式：BATCH和STREAMING。你通过FileSource.forBulkFileFormat()还是FileSource.forRecordStreamFormat()来选择，再配合Source的boundedness来区分是有限数据源还是无限数据源。

批模式下，FileSource 在初始化时做一次文件枚举，把目录下所有符合条件的文件都变成 split，然后就全部交给下游并行处理，不会再对外部目录做任何监控。流模式下，FileSource 有一个 FileEnumerator 周期性地扫描目录（默认发现间隔是scan-period，在StreamingFileSource的配置里可以调），新出现的文件会被包装成新的 split 提交给 SourceReader。

这里有个非常关键的点：流模式读取文件时，文件本身必须是“闭合状态”。什么算闭合状态？就是你把文件写完了，不会再往里追加数据了。Flink 怎么判断文件是否闭合？通过FileSystem上文件的时间戳、路径模式、以及文件大小是否长时间不变来综合判断。如果你用 Flink CDC 或者自研程序一边写文件一边让 FileSource 去读，大概率会把半截文件读进来，然后解析失败。

我当时的做法是：数据产出端先把文件写到tmp目录，写完后原子性地 move 到正式目录，正式目录下的文件名统一带_SUCCESS标记或者用.done文件做确认。FileSource 只监听正式目录，天然规避了“读到未完成文件”的问题。

2.2 通过 BulkFormat 读 Parquet 时，RowData 是怎么被“喂”出来的

DataStream API 里最常见的写法是：

FileSource<RowData> source = FileSource.forBulkFileFormat( new ParquetRowDataBuilder().createReaderFactory(schema), new Configuration() ).monitorContinuously(Duration.ofSeconds(10)) .build();

这里的ParquetRowDataBuilder并不是 Flink 官方提供的类，它只是我封装的一个工具，底层调用的是FlinkParquetReaders或者ParquetRowDataReader。官方 Flink 提供的 parquet 格式类在flink-parquet这个模块里，针对 Table/SQL 场景主要是ParquetFormatFactory，它会把 Parquet 文件映射成RowData内部结构。

如果你从 DataStream 直接走RowData，要注意 schema 的声明顺序必须跟 Parquet 文件里的字段顺序一致或能按名字对应上。FlinkParquetReaders是按列名去匹配的，不是按下标。所以我建议在解析 Parquet schema 时，先把它转换成 Flink 内部的RowType，然后传入 reader。

// 伪代码示意：从 Parquet 文件读出 schema，转换为 Flink RowType org.apache.parquet.schema.MessageType parquetSchema = ParquetFileReader.readFooter(...).getFileMetaData().getSchema(); RowType rowType = ParquetSchemaConverter.fromParquetType(parquetSchema, "record");

这里面 Parquet 的INT96类型极坑。Hive 老版本写 timestamp 经常用 INT96，Flink 的 parquet 读取器如果遇到 INT96，默认可以转换成 TIMESTAMP，但如果你的 jar 包用的 parquet-hadoop 版本比较老，或者 schema 转换逻辑没有显式处理 INT96，就会直接抛UnsupportedOperationException。我项目的解决方式是统一用TIMESTAMP(NANOSECONDS)类型声明，并在构建ParquetSchemaConverter时允许 INT96 映射到TIMESTAMP(9)。

2.3 批流一体的配置解剖：目录监听、文件过滤、读取超时

批流一体不是“同一份代码自动既跑批又跑流”，而是同一套 API 可以通过参数切换模式。把monitorContinuously去掉就是批，加上就是流。默认情况下：

• forBulkFileFormat(...).build()：批模式
• .monitorContinuously(interval)：流模式
• .processStaticFileSet()：只在任务启动时扫描一次，后续即使 monitorContinuously 加了，也只扫静态集合

这里有个容易被忽略的问题：当你用monitorContinuously时，SourceReader 读取完一批文件之后，会被挂起等待新的 split。如果你想同时设置“不读取隐藏文件”或者“只读某种前缀”，需要自定义FileEnumerator.Provider。

FileSource<RowData> source = FileSource.forBulkFileFormat( format, path ).monitorContinuously(Duration.ofSeconds(30)) .setFileEnumerator(new FileEnumerator.Provider() { @Override public FileEnumerator create(Configuration conf) { return new ContinuousFileEnumerator(...); } }) .build();

文件过滤逻辑藏在FileSystem的getFileStatus和listStatus里，默认情况下隐藏文件（以.开头）是不参与枚举的。另外，对象存储（OSS/S3）的场景下，listStatus有列表分页限制，文件一旦超过千级别，建议开启FileSystem的listStatus线程池，否则单单发现文件就可能花掉几十秒。

很多文章里提到“批流一体代码只需写一遍”，这话严格来说没错，前提是你要把“流的目录发现周期”和“批的全量枚举”都纳入设计。我之前参与的一个需求是：同一个数据管道，白天增量跑流（每 10 分钟扫描一次新文件），凌晨全量回溯跑批（把昨日全量目录一次性读入）。我的做法就是同一个 source builder，把monitorContinuously的参数从10s调成null（批）即可，下游代码完全不动。这个思路省掉了一套“批任务和流任务各自维护一套读取逻辑”的重复开发。

3. Avro 的三种打开方式：Generic、Specific、Reflect 该怎么选

3.1 三种模式到底在解决什么问题

Avro 作为序列化框架，最核心的资产是 schema。你在写数据时定义好 schema，读数据时只要拿到同一个 schema，不管数据在哪个节点上，都能正确反序列化。Flink 的flink-avro模块把 Avro 的三种数据处理模式都封装了，但很多人不清楚它们之间的边界。

• Generic（通用模式）：数据以GenericRecord形式读入，字段通过record.get("fieldName")访问。schema 是在运行时由外部传入的，不需要预先编译 Java 类。适合 schema 经常变、传输层不关心字段强类型的场景。
• Specific（特定模式）：用 Avro 的 schema 编译器生成 Java 类，每个字段映射成类的成员变量，类型安全，性能相对更好。适合 schema 基本稳定、团队内可以统一维护生成类的场景。
• Reflect（反射模式）：直接读取 Java POJO，利用反射推断 schema。适合已有大量 JavaBean、不想为了 Avro 单独生成类的场景，但要求字段类型和 Avro 类型能够自动对应，否则会出现奇怪的序列化结果。

在 Flink 里，当你用AvroDeserializationSchema或AvroRowDataDeserializationSchema时，底层就是在这三种模式里选。

3.2 Flink SQL 里读 Avro 时，RowData 和 GenericRecord 的桥接逻辑

如果你用的是 Flink SQL / Table API，直接通过WITH ('format' = 'avro')建表，底层会把 Avro schema 转换成 Flink RowType，然后创建AvroRowDataDeserializationSchema。在编译阶段，Flink 会把 Avro 的字符串类型映射成STRING，把bytes映射成BYTES，把fixed映射成BYTES，把enum映射成STRING，把union映射成字段可为空。

但这里有个经典误区：Avro 的 logicalType（比如 decimal、timestamp-millis）不会自动映射成正确的 Flink 类型。如果你只靠 SQL DDL 的 schema 声明而不在 avro schema 里显式标记 logicalType，读出来的 decimal 可能只是BYTES数组，timestamp 可能只是BIGINT。

我在项目里为了统一处理，写了一个 schema 转换工具类，核心逻辑就两件事：

1. 从 Avro schema 递归遍历每个字段。
2. 遇到logicalType时，根据类型名手动映射成 Flink 的DECIMAL(p,s)、TIMESTAMP(3)或TIMESTAMP_LTZ(3)。

// 示意：Avro logicalType 到 Flink dataType 的映射片段 if ("decimal".equals(logicalType)) { return DataTypes.DECIMAL(precision, scale); } else if ("timestamp-millis".equals(logicalType)) { return DataTypes.TIMESTAMP(3); } else if ("timestamp-micros".equals(logicalType)) { return DataTypes.TIMESTAMP(6); }

这个转换不能依赖 Flink 内置的AvroSchemaConverter，因为那个类在很多版本里对 timestamp 的默认映射是TIMESTAMP(3)，而你的 Parquet 里可能存的是TIMESTAMP(6)或者TIMESTAMP(9)。精度对不上，时间字段就会集体“丢精度”。

3.3 Specific 模式在 DataStream API 里的完整使用姿势

如果你已经用 Avro 编译器生成了UserProfile类，想用 DataStream API 直接读取包含 Avro 序列化数据的 Kafka topic，代码基本是：

DataStream<UserProfile> stream = env.addSource( new FlinkKafkaConsumer<>( "user-profile-topic", AvroDeserializationSchema.forSpecific(UserProfile.class), kafkaProps ) );

如果你想从文件（Parquet）里读出来直接是 Specific 类，事情就没那么简单了——Flink 的 parquet 格式插件目前没有直接暴露“反序列化成指定 POJO”的方法，默认就是 RowData。所以我当时做了一个折中：用 Parquet 读出 RowData，然后通过RowDataToAvroConverters转成 GenericRecord，再手动把 GenericRecord 转成 Specific 类。

GenericRecord generic = (GenericRecord) RowDataToAvroConverters.createConverter(avroSchema) .convert(rowData); UserProfile profile = SpecificData.get().deepCopy(UserProfile.SCHEMA$, generic);

这一套转换虽然在性能上不如直接读，但在逻辑上很稳，因为 RowData 和 Avro 之间的字段映射是可控的。

3.4 Reflect 模式实际跑起来有多“不靠谱”

Reflect 模式听名字很省事，我劝你别太当真。它有很强的隐式约定，比如：

• 字段类型必须是String、Integer、Long、Float、Double、Boolean、byte[]等基本类型，否则 Avro 会尝试用自定义编码，极其容易出错。
• 对象必须有 public 的无参构造函数，否则反序列化直接失败。
• 如果字段里有个List<LocalDateTime>，Avro 反射会直接报Unsupported operation，因为 Avro 的集合类型只认Collection，但LocalDateTime不是 Avro 原生类型。

所以 Reflect 模式我最终只用在“内部小数据量、字段极简、对象生命周期短”的场景里，生产链路绝对不用。

4. PyFlink 依赖与用户 JAR 的隐性冲突，常见错法及正解

4.1 PyFlink 到底是怎么跑起来的

PyFlink 不是“用 Python 重新实现一个 Flink”，它是在 Java 的 Flink 集群上通过PythonFunctionRunner启动了一个 Python 进程，Java 和 Python 之间通过数据交换协议（主要是 Beam 的FnApi）进行通信。这意味着：你的 Java/Scala 依赖（比如 parquet、avro、hadoop 客户端）是进 Java 侧的 classpath，Python 依赖是进 Python 侧的 environment。两者是两套体系。

这一点直接导致一种很普遍的困惑：你在 Python 侧pip install了一堆库，但 Java 侧读 Parquet 时可能仍然报“找不到 org.apache.parquet.avro.AvroParquetReader”。因为那是 Java 侧的类，跟 Python 环境没有任何关系。

正确理解应该是：

• PyFlink 的 Python 部分只负责 PyFlink 自身的 API 调用（TableEnvironment、DataStream API的 Python 封装）。
• Java 侧的 UDF、格式插件、Connector 全部跑在 Java 节点上，需要的依赖通过 JAR 打包进flink run的 classpath 或lib目录。

4.2 一个典型的错误做法：只给 Python 环境装依赖

我在某次对内分享时问过现场，某开发者的场景是：PyFlink 作业要读取 OSS 上的 Parquet 文件，他pip install pyarrow把 Parquet 读成 pandas DataFrame，然后转成 PyFlink Table。这个思路理论上可行，但一旦文件量大了，Python 侧轮询 OSS、逐文件读、再转 Table，性能就是灾难级。更关键的是，这不是“Flink 读 Parquet”，而是“Python 读 Parquet 再喂给 Flink”。

如果目标是用 Flink 的分布式能力去读，那 Parquet 解析必须发生在 Java TaskManager 上，必须给 Flink 集群提供flink-parquet、parquet-hadoop、hadoop-common这些 JAR。

4.3 把 Java 依赖塞进 PyFlink 作业的正确姿势

PyFlink 作业可以通过两种方式提供 Java 依赖：

1. 修改FLINK_HOME/lib目录：把所需的 JAR 放进去，重启集群。适合全局都要用的依赖。
2. 在 Python 脚本里用TableEnvironment.add_jars()或add_classpath()：只对当前作业生效。

我当时做的验证脚本长这样：

from pyflink.table import EnvironmentSettings, TableEnvironment env_settings = EnvironmentSettings.in_streaming_mode() t_env = TableEnvironment.create(env_settings) t_env.get_config().set("pipeline.classpaths", "hdfs:///jars/flink-parquet.jar," "hdfs:///jars/parquet-hadoop.jar," "hdfs:///jars/parquet-avro.jar")

注意一个坑：pipeline.classpaths配置在 PyFlink 里可能不生效，尤其是从flink run -py提交作业时，比较稳定的方式是直接add_jars：

t_env.add_jars("hdfs:///jars/flink-parquet.jar", "hdfs:///jars/parquet-avro.jar")

另外，如果你用 SQL Client 提交，也可以直接ADD JAR 'hdfs:///jars/xxx.jar';。

4.4 Distribute 依赖、Archives 和 Python UDF 的混用

PyFlink 里还经常见到这种场景：不仅需要 Java 侧读 Parquet，还需要 Python UDF 里对字段做处理。这时候你同时面对两套依赖：

• Java 侧：parquet、avro、flink-parquet。
• Python 侧：比如pyarrow、pandas、numpy。

Python 依赖分发通过TableEnvironment.add_python_archive()或者add_python_file()，也可以pip install之后用--pyfs传给集群。我踩过的一个坑是：Python UDF 里 import 了一个包，这个包依赖 C 库，直接pip install没问题，但打包成 zip 分发后，C 扩展的.so文件因为目录结构没对齐，Python import 不到。后续的解决方法是：用add_python_archive把整个site-packages目录打包，然后PYTHONPATH指到解压后的目录。

Java 侧和 Python 侧的依赖虽然互不干扰，但有一条线是打通的：PyFlink 的 UDF 最终是通过 Java 的PythonFunction算子调用的，Java 算子需要能找到 Python 环境的解释器。如果你的集群里 Java 进程启动 Python 子进程时找不到 Python home，就会报Python environment not found。这个错误跟依赖 JAR 无关，但排查时很容易被误判成“依赖没加全”。

5. 从业务事故里捞出来的三个排错案例

5.1 案例一：数据文件一切正常，但 Parquet Reader 频繁抛列违例

现象：同一个目录下的 Parquet 文件，用 Spark 读一切正常，用 Flink 读，任务跑不到几分钟就报错，错误堆栈指向ColumnarBatch的某个字段类型转换失败。

排查过程：

第一步，确认 Flink 侧声明的 schema。我发现建表语句里时间字段用的是TIMESTAMP(3)，但实际 Parquet 文件里的时间列在 Avro schema 里被标记为timestamp-micros，也就是微秒精度。Flink 的 ParquetRowDataReader 读 micros 时默认转换为TIMESTAMP(6)，往TIMESTAMP(3)里塞自然溢出了。

第二步，用parquet-tools查看文件 schema，确认文件里的列类型确实跟建表声明不一致。

修复方式：统一口径，建表时按文件内精度声明，下游需要TIMESTAMP(3)再通过CAST转换，而不是让 Reader 隐式截断。

5.2 案例二：流模式下 FileSource 监控目录，新文件永远读不到

现象：任务启动时，历史文件全部能读到，等 10 分钟后再往里丢新文件，始终读不到。

排查过程：

第一步，确认monitorContinuously确实生效了。日志里FileEnumerator一直在跑，说明目录监控没问题。

第二步，重点检查了文件命名规则。数据产出端用 Spark Streaming 写 OSS，文件名格式是part-xxxx-xxxx.c000.snappy.parquet。这个命名本身没有任何问题，但问题在于：作业用PathFilter只选择了以.parquet结尾的文件，新增文件在写入过程中会以临时名存在，比如part-xxxx-xxxx.c000.snappy.parquet.tmp，PathFilter把它过滤掉了。当临时文件被 rename 成正式文件名时，因为有AtomicRename语义，文件状态才被FileStatus捕获到。

听上去没问题，对吧？但如果数据产出端往同一个目录里重复生成同名文件（比如 task restart），Flink 在流模式下会把只处理“新增且从未处理过的文件”，同名文件如果modification time没有变化，直接视为已处理，新数据永远不会被读到。

解决办法：数据产出端 rename 时带上递增序号，比如part-0001.parquet、part-0002.parquet，确保文件名唯一。同时清理旧的临时文件，避免listStatus返回一堆垃圾状态导致主目录文件列表膨胀。

5.3 案例三：PyFlink 作业读了 Parquet，但 Python UDF 拿到的字段全部为空

现象：Java 侧用 SQL 建表读取 Parquet 成功，注册 Python UDF 后，字段传进来全是 null。

排查过程：

Python UDF 收到的数据其实是经过 Java 侧RowData序列化后的一段字节流，PyFlink 的PythonScalarFunctionRunner会先把 RowData 转成 Beam 的WindowedValue，再序列化传给 Python。这个序列化依赖flink-python里的一个类型序列化器。如果字段类型在 PyFlink 类型系统里找不到对应的 Python 类型，就会退化成 null。

我当时遇到的是DECIMAL(38,18)类型。Java 侧是正常的BigDecimal，但 PyFlink 的 Python 端把它转换成decimal.Decimal时，如果精度超过 Python 默认处理范围，可能被转成None。规避方案：Python UDF 的入参类型先改成 STRING，让 Java 侧转成字符串再传，Python 端再自行 parse。

这类问题表面上看是“UDF 写得有问题”，实际是类型序列化链路在传输层做了对不上的映射，排查时不要一上来就怀疑 UDF 逻辑，先打印一下func_type和input_types。

6. 一套可以直接抄走的生产级配置模板

6.1 Maven 依赖的组合和版本坑

先说明一下我在项目里稳定跑了一段时间的依赖组合（Flink 1.15 + Hadoop 3.3.1）：

这里有个大坑：flink-parquet 的 1.15.4 内部依赖的 parquet 版本是 1.12.3，但部分 flink-shaded-hadoop 的版本会传递依赖出不一致。如果 maven 依赖树里出现两个不同版本 parquet，那就极有可能在运行时抽风。解决方案是显式锁定 parquet 版本。

6.2 建表 DDL 和 FileSource 参数对照

如果要直接用 SQL 读 OSS 的 Parquet，DDL 长这样：

CREATE TABLE ods_user_log ( user_id BIGINT, event_time TIMESTAMP(6), event_type STRING, ext_data STRING, pt STRING ) PARTITIONED BY (pt) WITH ( 'connector' = 'filesystem', 'path' = 'oss://your-bucket/warehouse/user_log', 'format' = 'parquet', 'source.monitor-interval' = '10s', 'streaming-source.enable' = 'true', 'parquet.avro.read-flag' = 'true', 'parquet.avro.logical-timestamp-type' = 'TIMESTAMP_MICROS' );

注意parquet.avro.read-flag这个参数不是官方的，是我自定义的扩展配置，用来告诉我的自定义 factory 走 Avro 逻辑类型转换。官方默认的ParquetFormatFactory不认这个参数，如果你需要精确控制 micros 精度，最好自持一把DecodingFormat实现。

如果用 DataStream API，配置对照如下：

Configuration conf = new Configuration(); conf.setBoolean("flink.parquet.avro-compatible", true); conf.setString("flink.parquet.avro.logical-timestamp-type", "TIMESTAMP_MICROS"); FileSource<RowData> fileSource = FileSource.forBulkFileFormat( ParquetRowDataFormatFactory.create(schema, conf), monitorPath ).monitorContinuously(Duration.ofSeconds(30L)).build();

6.3 Python 侧严格适配 Java 依赖的顶层语法

当 PyFlink 和 Java 依赖必须共存时，我的目录结构长这样：

/opt/flink ├── lib/ # 集群级 JAR │ ├── flink-parquet-1.15.4.jar │ ├── parquet-hadoop-1.12.3.jar │ ├── parquet-avro-1.12.3.jar │ └── flink-avro-1.15.4.jar ├── pyflink-udf.zip # Python UDF 和 site-packages 打包 └── job.py # PyFlink 入口脚本

提交命令：

flink run \ -py /opt/flink/job.py \ -pyfs /opt/flink/pyflink-udf.zip \ -p 4 \ -c org.apache.flink.streaming.python.PythonStreamBinder

在 job.py 内部，记得先添加 Java JAR（如果没放到集群 lib 里）：

t_env.add_jars( "/opt/flink/lib/flink-parquet-1.15.4.jar", "/opt/flink/lib/parquet-hadoop-1.12.3.jar", "/opt/flink/lib/parquet-avro-1.12.3.jar", "/opt/flink/lib/flink-avro-1.15.4.jar" )

我个人的习惯是：能放集群 lib 的绝不塞作业里。因为add_jars在不同会话模式下有概率触发 classloader 不一致，一旦 TaskManager 上出现多个 classloader，同一个 parquet 类被不同 loader 加载，直接报IllegalAccessError。

7. 想清楚再动手：几件必须提前设计的决策

讲了这么多实现细节，最后从我个人的工程角度，再给你提几个可能在动手前就该想清楚的决策点。

第一，你到底是“读文件内容”还是“读文件作为流”。如果是前者，不要上 FileSource 的流模式，直接做定时批处理调度，逻辑简单得多，维护成本也低。FileSource 流模式适合“文件持续产生、需要准实时消费”的场景，但这时候你要牺牲掉“文件上游正在写”的容错，必须多一道“rename/确认文件”的环节。

第二，类型精度是你最大的敌人。Parquet 和 Avro 的 schema 如果没有在源头统一管理，下游排查的成本远高于字段本身带来的收益。我建议在数据产出端强制用同一个 schema 注册中心（或者至少一个共享的 Avro schema 文件库），任何字段类型变更都必须走版更流程，这样 Flink 侧只需要跟着升 schema 版本，而不用在每一层做兼容转换。

第三，PyFlink 的 Java 和 Python 依赖各自闭环，这句话值得在所有团队里反复强调。很多“PyFlink 作业莫名挂掉”的问题，根本不是 Flink 本身的问题，而是 Python 进程启动失败、或者 JAR 没进 Java classpath 导致的。排错时先快速判断“是 Java 侧崩了，还是 Python 侧崩了”，别上来就改代码。

第四，一条切身体会：用 DataStream API 读 Parquet 和 用 SQL 读 Parquet，在既有的 Flink 版本里还是两条不完全一致的路。SQL 的好处是简单、天然有 schema 声明；DataStream 的好处是灵活，但你必须自己维护RowType。如果你的团队都是 SQL 背景，建议直接把 DDL 建表方案作为主导；如果你的团队有比较重的 Scala/Java 数据工程背景，DataStream 方案反而排错更直接。两者没有绝对优劣，关键是统一团队内部的模式，别一半人用 SQL 一半人用 DataStream，出了问题互相看不懂代码。