PyFlink 1.17+ 连 Kafka 必须使用 flink-sql-connector-kafka-JAR，而非 kafka-python；JAR 需严格匹配版本并置于 lib/ 目录，否则报“Unknown connector identifier: kafka”。
PyFlink 1.17+ 连 Kafka 必须用 flink-sql-connector-kafka，不是 kafka-python
PyFlink 的 SQL API 和 DataStream API 都不直接依赖客户端库（如 kafka-python），而是通过 Flink 自带的 connector JAR 加载 Kafka 生产/消费逻辑。你 pip install 的 kafka-python 对 PyFlink 的 Kafka 表操作完全无效，装了也白装。
实操要点：
- 下载与 PyFlink 版本严格匹配的 connector JAR：
flink-sql-connector-kafka-<version>.jar</version>（例如 PyFlink 1.18 对应flink-sql-connector-kafka-1.18.1.jar） - JAR 必须放在
lib/目录下（启动 PyFlink 时通过--pyfiles或--jars指定无效，仅lib/被 classpath 扫描） - 不要手动修改
CLASSPATH：Flink 启动脚本会忽略它；改了也没用 - 验证是否加载成功：启动 job 后查日志，出现
Loaded connector factory for identifier 'kafka'才算到位
CREATE TABLE 语句里 'connector' = 'kafka' 但报 Unknown connector identifier: kafka
这是最典型的“JAR 没放对位置”错误。PyFlink 运行时找不到 connector 实现类，SQL 解析器就直接报这个错，而不是更具体的 ClassNotFoundException。
排查步骤：
立即学习“Python免费学习笔记（深入）”；
- 确认
lib/下有且仅有 1 个flink-sql-connector-kafka-*.jar（多个版本共存会导致 ServiceLoader 冲突） - 检查 JAR 内是否存在
META-INF/services/org.apache.flink.table.factories.Factory文件，且内容含org.apache.flink.connector.kafka.table.KafkaDynamicTableFactory - 用
jar -tf flink-sql-connector-kafka-1.18.1.jar | grep KafkaDynamicTableFactory快速验证 - 如果用的是 Confluent Schema Registry，额外需要
flink-sql-connector-kafka-base-*.jar（基础包，非可选）
Kafka 表定义中 'format' = 'json' 但消费时报 Failed to deserialize JSON
PyFlink 的 JSON format 默认要求字段名和类型与 DDL 完全一致，且不自动忽略多余字段或容忍空值——这和 Kafka CLI 或 kafka-python 的宽容行为完全不同。
常见踩坑点：
-
json.ignore-parse-errors = 'true'必须显式设置，否则任意字段类型错（比如 string 写成 int）、字段缺失、JSON 格式损坏都会中断整个作业 - 时间字段必须用
TIMESTAMP_LTZ并配'json.timestamp-format.standard' = 'ISO-8601'，否则"2024-05-20T10:30:00Z"会被当成字符串 - 嵌套 JSON 不被原生支持：DDL 不能写
data ROW<user_id bigint profile row int>></user_id>，得先用json格式读成 STRING，再用JSON_VALUE或 UDF 提取 - 注意时区：Kafka broker 返回的时间戳是 UTC，但 Flink 本地 session timezone 可能是系统默认，导致
PROCTIME()和事件时间对不上
PyFlink 作业提交后卡在 DEPLOYING 状态，日志无明显报错
这通常是 Kafka 连接参数配置错误，但 Flink 不会在初始化阶段抛异常，而是后台不断重试直到超时（默认 60 秒），期间状态卡住。
关键检查项：
-
'properties.bootstrap.servers'值末尾不能带空格，也不能用localhost（容器环境需用宿主机 IP 或服务名） -
'properties.group.id'必须设置（即使是 source-only 作业），否则 KafkaConsumer 构造失败但静默吞掉异常 - SSL 认证场景下，
'properties.security.protocol' = 'SSL'必须配合'properties.ssl.truststore.location'等完整路径（注意是 JobManager 节点上的路径，不是本地开发机） - 若 Kafka 启用了 SASL/PLAIN，
'properties.sasl.jaas.config'的值必须是完整 JAAS 字符串，不能只写文件路径
真正麻烦的是这些配置错误不会立刻暴露——得去 TaskManager 日志里搜 org.apache.kafka.clients.NetworkClient 或 Unable to connect to 才能看到底层连接拒绝细节。
