Flink写入S3时提示找不到Orc格式工厂的问题求助

我尝试将Flink表结果写入S3的ORC文件，代码实现如下：

tEnv.createTemporaryTable("my_output_table", TableDescriptor.forConnector("filesystem")
                .schema(outputSchema)
                .option("path", s3OutputPath)
                .format(FormatDescriptor.forFormat("orc").build())
                .build());
finalResultToInsert.executeInsert("my_output_table");

运行时抛出如下错误：

Caused by: org.apache.flink.table.api.ValidationException: Could not find any format factory for identifier 'orc' in the classpath.
    at org.apache.flink.table.filesystem.FileSystemTableSink.<init>(FileSystemTableSink.java:128) ~[flink-table_2.12-1.14.2.jar:1.14.2]
    at org.apache.flink.table.filesystem.FileSystemTableFactory.createDynamicTableSink(FileSystemTableFactory.java:87) ~[flink-table_2.12-1.14.2.jar:1.14.2]
    at org.apache.flink.table.factories.FactoryUtil.createTableSink(FactoryUtil.java:179) ~[flink-table_2.12-1.14.2.jar:1.14.2]
    at org.apache.flink.table.planner.delegation.PlannerBase.getTableSink(PlannerBase.scala:394) ~[flink-table_2.12-1.14.2.jar:1.14.2]
......

我已经引入了相关依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-orc_2.11</artifactId>
    <version>1.14.2</version>
</dependency>

生成的Jar包中也包含ORC相关类，但依然提示找不到格式工厂。项目中的Flink-Avro依赖可以正常使用，运行环境是AWS EMR 6.6.0（内置Flink 1.14.2），求解决方法。

1. 修正Scala版本匹配问题 #

错误日志显示Flink核心包基于Scala 2.12编译（flink-table_2.12-1.14.2.jar），但你引入的ORC依赖是Scala 2.11版本，版本不匹配会导致类加载失败。将依赖修改为Scala 2.12版本：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-orc_2.12</artifactId>
    <version>1.14.2</version>
</dependency>

2. 确保SPI配置文件被正确打包 #

Flink通过Java SPI机制加载格式工厂，需要保证META-INF/services/org.apache.flink.table.factories.Factory文件被正确合并到最终Jar包中。如果使用Maven Shade插件打包，需添加服务配置转换：

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>3.2.4</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                    <configuration>
                        <transformers>
                            <!-- 合并SPI服务配置文件 -->
                            <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                        </transformers>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

3. 调整EMR集群类加载优先级 #

EMR集群自带的Flink依赖可能和你的Jar包依赖冲突，提交作业时可以指定类加载顺序优先加载作业Jar中的类：

flink run -yD classloader.resolve-order=child-first your-job.jar

4. 显式配置ORC格式参数（可选） #

尝试显式添加ORC相关配置，确保格式被正确识别：

tEnv.createTemporaryTable("my_output_table", TableDescriptor.forConnector("filesystem")
        .schema(outputSchema)
        .option("path", s3OutputPath)
        .format(FormatDescriptor.forFormat("orc")
                .option("orc.compress", "SNAPPY") // 指定压缩方式，可选
                .build())
        .build());

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