Apache Flink SQL中凭据与其他密钥的合理存储方案咨询

Flink 原生提供3种无需二次开发的凭据传递方案，完全可以满足避免硬编码的需求，无需自定义UDF实现：

1. 原生占位符语法（支持读取环境变量、配置项、系统属性） #

Flink 1.13+ 版本原生支持${前缀.变量名}格式的占位符，可以直接在DDL的WITH参数中使用：

• 读取系统环境变量：使用${env.变量名}格式
• 读取flink-conf.yaml配置项：使用${config.配置键名}格式
• 读取JVM系统属性：使用${system.属性名}格式

使用示例：
你可以先在Flink进程运行环境中预先注入环境变量：

export PG_USER=实际用户名
export PG_PWD=实际密码

之后DDL语句可以直接引用，不需要写入明文凭据：

CREATE CATALOG analytics WITH (
    'type'='jdbc',
    'base-url'='jdbc:postgresql://some-postgres:5432/',
    'default-database'='analytics',
    'username'='${env.PG_USER}',
    'password'='${env.PG_PWD}'
);

2. 挂载密钥文件直接读取 #

Flink 1.15+ 版本额外支持file:前缀的占位符，可以自动读取指定路径下的文件内容作为配置值，适合密钥挂载场景（比如K8s Secret挂载、加密存储卷挂载）：

CREATE CATALOG analytics WITH (
    'type'='jdbc',
    'base-url'='jdbc:postgresql://some-postgres:5432/',
    'default-database'='analytics',
    'username'='${file:/var/secret/pg_username}',
    'password'='${file:/var/secret/pg_password}'
);

Flink会自动读取对应路径的文件内容作为参数值，不需要额外代码处理。

3. SQL Client 启动动态传参 #

如果使用SQL Client提交任务，你可以在启动命令中通过-D参数传入敏感配置，避免写入SQL文件：
启动命令示例：

./sql-client.sh -Dpg.user=实际用户名 -Dpg.pwd=实际密码

DDL引用方式：

CREATE CATALOG analytics WITH (
    'type'='jdbc',
    'base-url'='jdbc:postgresql://some-postgres:5432/',
    'default-database'='analytics',
    'username'='${config.pg.user}',
    'password'='${config.pg.pwd}'
);

Kafka 场景适配 #

以上所有方案完全适配Kafka表的凭据配置，示例如下：

CREATE TABLE kafka_source (
    id INT,
    name STRING
) WITH (
    'connector' = 'kafka',
    'topic' = '业务topic',
    'properties.bootstrap.servers' = 'kafka集群地址',
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'properties.sasl.mechanism' = 'PLAIN',
    'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username="${env.KAFKA_USER}" password="${env.KAFKA_PWD}";',
    'format' = 'json'
);

注意事项 #

• Flink默认会对所有敏感配置项做日志、UI展示的脱敏处理，不会泄露传入的凭据信息
• 占位符功能最低要求Flink 1.13版本，文件读取功能最低要求Flink 1.15版本，低于对应版本可以通过升级小版本适配，无需自定义开发

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