Apache Flink 1.15.2本地PyFlink作业Kinesis连接器类路径错误求助

以下是针对问题的具体排查和解决步骤：

• 匹配Kinesis连接器与Flink版本
  你使用的是Flink Table API，必须使用对应版本的Flink SQL Kinesis连接器jar，而非普通的Kinesis连接器jar。需下载flink-sql-connector-kinesis_2.12-1.15.2.jar（注意Scala版本要与你的Flink安装一致，通常为2.12），该jar包内置了Table API所需的DynamicTableFactory实现类。

• 确保jar被Flink正确加载
  本地运行PyFlink作业时，Flink默认优先读取安装目录（FLINK_HOME）下的lib文件夹，而非作业目录的lib子目录。你有两种选择：

  1. 将连接器jar复制到<FLINK_HOME>/lib目录，重启Flink本地集群（若使用集群模式）；
  2. 提交作业时通过命令行参数指定jar路径，示例命令：

     flink run -py getting-started.py --jarfile ./lib/flink-sql-connector-kinesis_2.12-1.15.2.jar
     

• 验证jar包完整性
  检查下载的jar包是否包含目标类，执行以下命令：

  jar tf flink-sql-connector-kinesis_2.12-1.15.2.jar | grep KinesisDynamicTableFactory
  

  如果无输出，说明jar包损坏或版本错误，需重新下载对应版本的jar。

• 确认PyFlink与Flink版本一致性
  在conda虚拟环境中执行pip show apache-flink，确保输出的版本是1.15.2，与本地安装的Flink版本完全匹配。版本不匹配会导致类加载冲突，引发工厂类找不到的错误。

• 显式设置作业Classpath
  如果上述方法无效，可在Python代码中显式添加jar依赖：

  from pyflink.datastream import StreamExecutionEnvironment
  
  env = StreamExecutionEnvironment.get_execution_environment()
  # 替换为你的jar文件绝对路径
  env.add_jars("file:///absolute/path/to/flink-sql-connector-kinesis_2.12-1.15.2.jar")
  

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