使用作业变量和密钥_实时计算 Flink版(Flink)-阿里云帮助中心

Flink支持使用作业变量和密钥，可以避免明文AccessKey、密码等信息带来的安全风险，通过复用可以避免重复编写相同的代码或值，易于配置管理。支持在SQL作业、JAR或Python作业、日志配置和UI界面等场景中使用。本文为您介绍如何创建变量和密钥，及对应的使用方法。

背景信息

作业变量和密钥都由名（称）和值组成，名是用来代替真实数据的名称，由您自定义，通常能够直接反映值的含义，值则是实际要使用而被名称代替展示的值。

作业变量：仅支持在对应的单个作业下使用，包括SQL作业的DDL、JAR/Python作业Main函数入参，使用格式为${变量名}，具体使用方法请参见作业变量。
密钥：支持在该项目空间下使用，包括SQL作业的DDL、日志配置和UI界面（创建Catalog），使用格式为${secret_values.密钥名}，具体创建和使用方法请参见密钥托管。

注意事项

仅项目空间中具有editor或owner权限的成员可以配置和使用变量和密钥，授权操作详情请参见项目空间授权。
不支持为一个变量名或密钥名配置多个值。
SQL作业中仅支持在临时表（CREATE TEMPORARY TABLE）中使用变量和密钥，持久表（CREATE TABLE）中使用会导致该表无法正常使用。

作业变量

说明

作业变量名需要在单个作业内保持唯一。

SQL作业

支持SQL作业开发中使用变量，并在作业启动时可再次修改变量值。

单击Flink开发控制台左侧导航栏的SQL开发。

在目标作业草稿的SQL编辑器中使用${变量名}格式创建变量，示例如下。

在WITH参数中使用时，需要使用单引号，'${变量名}'。
在其他位置使用（例如SQL表名、字段）时，需要使用反引号`${变量名}`。

create temporary table `${source_name}`(
    id varchar,
    name varchar
) with (
    'connector' = 'datagen'
);


create temporary table blackhole(
    id varchar,
    `${test_name}` varchar
) with (
    'connector' = '${blackhole}'
);

insert into blackhole select * from `${source_name}`;

Flink会自动识别作业草稿中使用了该格式的作业变量，您需要填写变量值。
- 方法一：SQL编辑器会出现作业变量面板，可直接在面板中进行填写。
- 方法二：单击右侧的更多配置，在作业变量区域进行填写。
作业部署后，您可以在作业运维页面目标作业的部署详情页签查看作业使用的变量。
单击目标作业操作列下的启动。支持在作业启动面板中修改变量值。
重要
作业启动面板中修改变量值仅对作业本次启动生效，不会修改SQL草稿中对应的变量值。

JAR/Python作业

支持JAR和Python作业中Main函数入参使用变量。

在开发控制台的作业运维页面，单击部署作业。
在Entry Point Main Arguments参数位置输入入参变量名，使用格式如下，并在参数下方的作业变量部分填写变量值。
其他部署参数配置详情请参见部署作业。
单击部署。
单击目标作业操作列下的启动。支持在作业启动面板中修改变量值。

密钥托管

说明

密钥名需在同一个项目空间下保持唯一。
仅支持新增和删除密钥，不支持修改和查看密钥值。
请谨慎删除密钥，可能会导致使用变量的作业无法部署，或需要重新部署已上线的作业。但不影响运行中的作业。

新增密钥

登录实时计算控制台，单击目标工作空间操作列下的控制台。
在安全中心页面，单击密钥托管页签。
单击新增密钥，填写密钥名称和密钥值。
重要
密钥值对字母大小写敏感，填写时请注意字母大小写。
单击确定。

使用密钥

增加密钥后，您可以通过${secret_values.密钥名}格式使用，使用时您仅需将其中的密钥名进行修改，其他部分保留不变，具体示例如下。

SQL作业

支持在SQL作业的DDL中通过${secret_values.密钥名}格式使用密钥，示例如下。

在WITH参数中使用时，需要使用单引号，'${secret_values.密钥名}'。
在其他位置使用（例如SQL表名、字段）时，需要使用反引号`${secret_values.密钥名}`。

create temporary table `${secret_values.source_name}`(
    id varchar,
    name varchar
) with (
    'connector' = 'datagen'
);


create temporary table blackhole(
    id varchar,
    `${secret_values.test_name}` varchar
) with (
    'connector' = '${secret_values.blackhole}'
);

insert into blackhole select * from `${secret_values.source_name}`;

日志配置

进入日志配置页面。
- 配置单个作业，具体操作请参见配置单个作业日志输出。
- 配置项目空间下所有作业，具体操作请参见配置项目空间下所有作业日志输出。

在日志配置中使用密钥。

例如，在配置作业日志输出到SLS时使用名为accessKeyId和accessKeySecret的密钥示例如下，完整日志输出配置信息具体请参见配置作业日志输出。

<Appender name="SLS" type="SLS">
  <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/>  

  <!-- The final effective log path is: ${baseUri}/logs/${namespace}/${deploymentId}/{jobId}/ -->
  <Property name="namespace">{{ namespace }}</Property> <!-- Do not modify this line -->
  <Property name="project">YOUR-SLS-PROJECT</Property>  
  <Property name="logStore">YOUR-SLS-LOGSTORE</Property> 
  <Property name="endpoint">YOUR-SLS-ENDPOINT</Property> 
  <Property name="accessKeyId">${secret_values.accessKeyId}</Property> 
  <Property name="accessKeySecret">${secret_values.accessKeySecret}</Property> 
  <Property name="topic">{{ namespace }}:{{ deploymentId }}:{{ jobId }}</Property>
  <Property name="deploymentName">{{ deploymentName }}</Property>
  <Property name="flushIntervalSeconds">10</Property>
  <Property name="flushIntervalEventCount">100</Property>
</Appender>

UI界面

在通过UI界面方式创建Catalog时，您可以在配置项中使用密钥。

在Flink开发控制台左侧导航栏，单击元数据管理。
在创建Catalog页面使用密钥。例如在MySQL Catalog中的password参数使用名为mysqlpassword的密钥示例如下。

变量和密钥管理

背景信息

注意事项

作业变量

SQL作业

JAR/Python作业

密钥托管

新增密钥

使用密钥

SQL作业

日志配置

UI界面

相关文档