Docker环境下PyFlink执行execute_sql触发ruamel.yaml解析错误

以下是针对该问题的排查和解决方案：

1. 排查DDL中WITH子句的YAML语法 #

PyFlink解析SQL的WITH参数时会用YAML语法处理配置项，任何语法错误（比如引号不闭合、冒号缺失、缩进错误）都会触发解析异常：

• 把DDL中WITH块的配置单独提取出来，用ruamel.yaml做本地测试：

  import ruamel.yaml
  # 替换成你DDL里的WITH配置内容
  yaml_config = """
  properties.bootstrap.servers: kafka:9092
  format: json
  scan.startup.mode: earliest-offset
  """
  try:
      ruamel.yaml.load(yaml_config, Loader=ruamel.yaml.Loader)
      print("YAML语法正常")
  except Exception as e:
      print(f"YAML语法错误: {e}")
  

• 确保WITH中的键值对格式正确，避免使用全角符号、多余逗号，所有字符串引号保持统一（单/双引号不要混用）。

2. 锁定ruamel.yaml兼容版本 #

Flink 1.19.0对ruamel.yaml版本有依赖要求，过高或过低版本都可能引发兼容性问题：

• 在自定义Docker镜像中指定安装与PyFlink 1.19.0兼容的版本，比如：

  RUN pip install --no-cache-dir ruamel.yaml==0.17.32
  

  该版本是Flink 1.19.x官方依赖的稳定版本。

3. 清理DDL中的隐藏特殊字符 #

复制粘贴的DDL可能带有不可见特殊字符（比如全角空格、异常换行符），导致YAML解析失败：

• 将DDL内容复制到纯文本编辑器（如VS Code、Notepad++），开启"显示所有字符"功能，检查并移除异常字符后重新编写DDL。

4. 用Table API替代execute_sql做对比验证 #

既然官方非SQL方式的示例正常，可尝试用Table API定义源表，验证是否能正常运行：

# 用Table API创建Kafka源表
tbl_env.connect(
    "kafka://kafka:9092",
    topic="your_input_topic"
).with_format("json").with_schema(
    "id INT, name STRING, ts TIMESTAMP(3)"
).create_temporary_table("kafka_source")

如果此方式正常运行，说明问题确实出在DDL的YAML解析环节，需重点排查DDL语法。

5. 检查Flink配置文件的YAML语法 #

自定义镜像中的flink-conf.yaml若存在语法错误，也可能间接导致SQL执行时的YAML解析异常：

• 检查flink-conf.yaml中的配置项，确保所有键值对都有正确的冒号分隔，缩进统一为2个空格，无语法错误。

内容的提问来源于stack exchange，提问作者Mateus P. Lourenço