Flink批处理TableEnvironment初始化报错：找不到ExecutorFactory的解决方法

错误原因 #

异常Could not find any factories that implement 'org.apache.flink.table.delegation.ExecutorFactory'的核心原因是类路径中缺少Flink Table批处理模式所需的执行器依赖，同时你的代码混用ExecutionEnvironment和TableEnvironment，存在环境管理冲突。

解决方案 #

1. 添加正确的依赖 #

根据你的Flink版本，添加对应的批处理执行器依赖：

• Flink 1.15及以上版本（推荐）：添加flink-table-planner-loader依赖

  <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-planner-loader</artifactId>
      <version>你的Flink版本号</version>
  </dependency>
  

• Flink 1.14及以下版本：添加flink-table-planner-blink依赖（注意Scala版本后缀）

  <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-planner-blink_2.12</artifactId>
      <version>你的Flink版本号</version>
  </dependency>
  

同时确保已添加flink-table-api-java依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java</artifactId>
    <version>你的Flink版本号</version>
</dependency>

2. 修正代码中的环境冲突 #

TableEnvironment在批模式下已经封装了执行环境，无需单独创建ExecutionEnvironment，也不需要调用env.execute()，直接通过TableEnvironment的方法触发任务执行：

// 移除原有的ExecutionEnvironment创建代码
TableEnvironment tenv = TableEnvironment.create(EnvironmentSettings.inBatchMode());

tenv.executeSql("CREATE TABLE Flinkdata (" +
        "  Inde STRING," +
        "  User_Id STRING," +
        "  First_Name STRING," +
        "  Last_Name STRING," +
        "  Sex STRING," +
        "  Email STRING," +
        "  Phone STRING," +
        "  Date_of_birth STRING," +
        "  Job_Title STRING," +
        "  PRIMARY KEY (Inde) NOT ENFORCED" +
        ") WITH (" +
        "   'connector.type' = 'jdbc'," +
        "   'connector.url' = 'jdbc:mysql://localhost/ruby'," +
        "   'connector.table' = 'flink_people_data'," +
        "   'connector.username' = 'root'," +
        "   'connector.password' = 'passwordd1234'" +
        ")");

// 执行查询并打印结果
tenv.executeSql("SELECT * FROM Flinkdata").print();
Table transactions = tenv.from("Flinkdata");

// 若需执行写入等操作，使用Table的executeInsert方法
// transactions.executeInsert("目标表名");

3. 检查版本一致性 #

确保所有Flink相关依赖的版本完全一致，避免版本冲突导致的类加载异常。

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