Flink SQL Client 如何支持 Kafka？Could not find any factory for identifier kafka / upsert-kafka 解决方法_could not find any factory for identifier 'kafka

我们以 EMR 为例，Flink SQL Client 默认是没有安装 kafka connector 的相关 Jar 包的，所以在 Flink SQL Client 中执行 Kafka 的建表语句时会报如下错误：

org.apache.flink.table.api.ValidationException: Could not find any factory for identifier ‘upsert-kafka’ that implements ‘org.apache.flink.table.factories.DynamicTableFactory’ in the classpath.

解决方法就是在 ${FLINK_HOME}/lib 上安装 Kafka Connector 的 Jar 包，我们以 AWS EMR 6.15 ( Flink 1.17.1 ) 版本为例，执行如下脚本即可：

# only run on master node is enough, owner of flink home dir is `flink` user
sudo -u flink wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/1.17.1/flink-sql-connector-kafka-1.17.1.jar -P /usr/lib/flink/lib/
1
2

注意：在 Yarn 模式下，Flink 只是一个 Yarn 的 App，EMR 的 Worker 节点上并没有 /usr/lib/flink 这个目录，也不存在将 Jar 包分发到所有 Worker 节点的 /usr/lib/flink 目录下这一说！这与 Spark 更类似，与部署 HBase Coprocessor Jar 包相反！