如何通过Flink Table API向远程集群提交SQL及排查依赖问题

一、通过Table API提交SQL到远程Flink服务器 #

可通过构建远程TableEnvironment来提交SQL任务，核心是配置集群连接信息并依次执行SQL语句，具体步骤如下：

• 添加必要依赖
  在Java Web项目的pom.xml中添加Flink Table API相关依赖（版本需与远程Flink服务器完全一致，以下以1.15.2为例）：

  <dependencies>
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-table-api-java-bridge</artifactId>
          <version>1.15.2</version>
          <scope>provided</scope>
      </dependency>
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-streaming-java</artifactId>
          <version>1.15.2</version>
          <scope>provided</scope>
      </dependency>
      <!-- Kafka连接器依赖：若集群已部署，此处设为provided避免冲突 -->
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-connector-kafka</artifactId>
          <version>1.15.2</version>
          <scope>provided</scope>
      </dependency>
  </dependencies>
  

• 构建远程TableEnvironment并提交SQL
  配置集群连接参数，初始化远程执行环境后依次执行源表创建、汇表创建、插入查询语句：

  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import org.apache.flink.table.api.EnvironmentSettings;
  import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  
  public class RemoteFlinkSqlSubmitter {
      public static void submitSqlToRemoteCluster() throws Exception {
          // 远程JobManager地址与端口（可使用Rest端口8081或RPC端口6123）
          String jobManagerHost = "your-flink-jobmanager-host";
          int jobManagerPort = 8081;
  
          // 初始化流式TableEnvironment配置
          EnvironmentSettings settings = EnvironmentSettings.newInstance()
                  .inStreamingMode()
                  .build();
  
          // 创建远程StreamExecutionEnvironment
          StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(
                  jobManagerHost,
                  jobManagerPort,
                  // 若需显式上传用户jar，在此添加路径数组，如new String[]{"path/to/connector.jar"}
                  new String[]{}
          );
  
          // 绑定TableEnvironment
          StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
  
          // 执行源表创建SQL
          String createSourceSql = "CREATE TABLE kafka_source (" +
                  "  id INT," +
                  "  name STRING" +
                  ") WITH (" +
                  "  'connector' = 'kafka'," +
                  "  'topic' = 'test_topic'," +
                  "  'properties.bootstrap.servers' = 'kafka-host:9092'," +
                  "  'format' = 'json'," +
                  "  'scan.startup.mode' = 'latest-offset'" +
                  ")";
          tableEnv.executeSql(createSourceSql);
  
          // 执行汇表创建SQL
          String createSinkSql = "CREATE TABLE print_sink (" +
                  "  id INT," +
                  "  name STRING" +
                  ") WITH (" +
                  "  'connector' = 'print'" +
                  ")";
          tableEnv.executeSql(createSinkSql);
  
          // 执行插入查询并提交任务
          String insertSql = "INSERT INTO print_sink SELECT id, name FROM kafka_source";
          tableEnv.executeSql(insertSql).await();
      }
  }
  

二、RemoteStreamEnvironment依赖加载问题分析 #

你遇到的"集群/lib目录已放依赖但报错，显式指定路径则正常"的问题，通常由以下原因导致：

• 集群未重启，依赖未生效
  Flink进程仅在启动时加载/lib目录下的jar包。若在集群运行后才添加依赖，必须重启JobManager和所有TaskManager，否则进程无法识别新jar。

• 集群节点依赖未同步
  若为Standalone或YARN集群，仅在JobManager的/lib放依赖不足：

  • Standalone集群：所有TaskManager节点的/lib目录需同步放置相同依赖，因为TaskManager执行任务时会加载这些类。
  • YARN集群：默认JobManager和TaskManager的lib从Flink分发包复制，若手动修改节点lib，需重新打包分发包或通过yarn.provided.lib.dirs配置指定依赖路径。

• 版本不匹配或类加载冲突

  • 客户端项目的Flink相关依赖版本必须与集群完全一致，否则会出现NoClassDefFoundError或IllegalAccessError。
  • 显式指定jar路径时，客户端会将jar上传到集群，此时集群优先加载用户上传的jar，避免了客户端与集群依赖版本不一致的冲突。

• 依赖包权限问题
  检查集群/lib目录下jar包的权限，确保Flink进程运行用户（通常为flink用户）拥有读权限，权限不足会导致进程无法读取类文件。

• 间接依赖缺失
  部分依赖（如kafka-clients）需要lz4-java、snappy-java等间接依赖，仅放置三个核心jar会导致类加载失败；而显式指定路径时，客户端可能通过Maven依赖传递将间接依赖一起上传，从而解决缺失问题。

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