Flink Jar使用DEW获取访问凭证读写OBS

操作场景

AI DataLake将Flink Jar作业的输出数据写入到OBS时，需要配置AK/SK访问OBS，为了确保AK/SK数据安全，您可以通过密码安全中心（Data Encryption Workshop，DEW）、云凭据管理服务（Cloud Secret Management Service，CSMS），对AK/SK统一管理，有效避免程序硬编码或明文配置等问题导致的敏感信息泄露以及权限失控带来的业务风险。

本例以获取访问OBS的AK/SK为例介绍Flink Jar使用DEW获取访问凭证读写OBS的操作指导。

约束与限制

仅支持Flink 1.20版本使用DEW管理访问凭据，在创建作业时，请配置作业使用Flink 1.20版本、且已在作业中配置允许AI DataLake访问DEW的委托信息。

前提条件

已在DEW服务创建通用凭证，并存入凭据值。具体操作请参考创建通用凭据。
使用该功能，所有涉及OBS的桶，都需要进行配置AK/SK。
已创建AI DataLake访问DEW的委托并完成委托授权。该委托需具备以下权限：
- DEW中的查询凭据的版本与凭据值ShowSecretVersion接口权限，csms:secretVersion:get。
- DEW中的查询凭据的版本列表ListSecretVersions接口权限，csms:secretVersion:list。
- DEW解密凭据的权限，kms:dek:decrypt。
委托权限示例请参考创建自定义委托和常见场景的委托权限策略。

语法格式

在用创建Flink作业接口提交作业时，在flink_config中配置DEW访问凭证，配置信息如下:

全局配置（对所有OBS桶生效）：

     flink.hadoop.fs.obs.dew.access.key=USER_AK_CSMS_KEY
flink.hadoop.fs.obs.dew.secret.key=USER_SK_CSMS_KEY
flink.hadoop.fs.obs.dew.session.token=USER_TOKEN_CSMS_KEY
flink.hadoop.fs.obs.security.provider=com.aidatalake.flink.auth.provider.DewObsCredentialProvider
flink.hadoop.fs.dew.csms.secretName=CredentialName
flink.hadoop.fs.dew.csms.version=VERSION_ID
flink.hadoop.fs.dew.csms.cache.time.second=CACHE_TIME
 
 
  

按桶配置（仅对指定桶生效，不同的OBS桶可使用不同的AK/SK认证信息）：

     flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.access.key=USER_AK_CSMS_KEY
flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.secret.key=USER_SK_CSMS_KEY
flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.session.token=USER_TOKEN_CSMS_KEY
flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.security.provider=com.aidatalake.flink.auth.provider.DewObsCredentialProvider
flink.hadoop.fs.dew.csms.secretName=CredentialName
flink.hadoop.fs.dew.csms.version=VERSION_ID
flink.hadoop.fs.dew.csms.cache.time.second=CACHE_TIME
 
 
  

按桶配置优先于全局配置：访问某个OBS桶时，若该桶配置了按桶配置，则优先使用按桶配置，未按桶配置的参数使用全局配置。

按桶配置对访问该桶的所有OBS操作生效，若作业Jar包及依赖文件存放在该桶，拉取作业文件时同样使用该桶对应的凭据，需确保凭据具备相应读取权限。

flink.hadoop.fs.dew.csms开头的参数仅支持全局配置。USER_BUCKET_NAME需替换为用户的OBS桶名，参数说明详见表1。

说明：DEW服务的endpoint和项目ID由AI DataLake平台在作业运行时自动注入，无需用户配置。作业通过已配置的作业委托的临时凭证访问DEW，因此该委托需具备查询CSMS凭据的权限。

参数说明

表1 参数说明
参数	是否必选	默认值	数据类型	参数说明
flink.hadoop.fs.obs.dew.access.key	是	无	String	参数支持两种配置形式：全局配置flink.hadoop.fs.obs.dew.access.key（对所有OBS桶生效），按桶配置flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.access.key（仅对该桶生效，USER_BUCKET_NAME需替换为用户的OBS桶名）。按桶配置优先于全局配置。参数的值为用户定义在CSMS通用凭证中的键key，其Key对应的value为用户的AK（Access Key Id），需要具备访问OBS对应桶的权限。
flink.hadoop.fs.obs.bucket.dew.secret.key	是	无	String	参数支持两种配置形式：全局配置flink.hadoop.fs.obs.dew.secret.key（对所有OBS桶生效），按桶配置flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.secret.key（仅对该桶生效，USER_BUCKET_NAME需替换为用户的OBS桶名）。按桶配置优先于全局配置。参数的值为用户定义在CSMS通用凭证中的键key，其Key对应的value为用户的SK（Secret Access Key），需要具备访问OBS对应桶的权限。
flink.hadoop.fs.obs.bucket.dew.session.token	否	无	String	参数支持两种配置形式：全局配置flink.hadoop.fs.obs.dew.session.token（对所有OBS桶生效），按桶配置flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.session.token（仅对该桶生效，USER_BUCKET_NAME需替换为用户的OBS桶名）。按桶配置优先于全局配置。参数的值为用户定义在CSMS通用凭证中的键key，其Key对应的value为用户的临时访问凭证securityToken。当存入通用凭证的AK/SK为临时凭证时，需要配置该参数；使用永久AK/SK时，无需配置。
flink.hadoop.fs.obs.security.provider	是	无	String	OBS AK/SK认证机制，使用DEW服务中的CSMS凭证管理，获取OBS的AK、SK。使用DEW方式访问OBS时，固定取值为com.aidatalake.flink.auth.provider.DewObsCredentialProvider。参数支持两种配置形式：全局配置flink.hadoop.fs.obs.security.provider（对所有OBS桶生效），按桶配置flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.security.provider（仅对该桶生效，USER_BUCKET_NAME需替换为用户的OBS桶名）。按桶配置优先于全局配置。不配置该参数时，默认使用作业委托的临时凭证访问OBS。
flink.hadoop.fs.dew.csms.secretName	是	无	String	在DEW服务的凭据管理中新建的通用凭据的名称。配置示例：flink.hadoop.fs.dew.csms.secretName=secretInfo
flink.hadoop.fs.dew.csms.cache.time.second	否	3600	Long	Flink作业访问获取CSMS通用凭证后，缓存的时间。单位为秒。默认值为3600秒。该参数必须为非负整数，否则作业启动失败。配置为0时，每次访问OBS都会重新从CSMS获取凭据。

样例代码

本章节Java样例代码演示将DataGen数据处理后写入到OBS，具体参数配置请根据实际环境修改。

创建AI DataLake访问DEW的委托并完成委托授权。详细步骤请参考创建自定义委托。
在DEW创建通用凭证。操作请参考创建通用凭据。
凭据值以JSON格式保存OBS的AK/SK信息，键名可自定义，需与优化参数中dew.access.key、dew.secret.key的取值保持一致。示例如下：
```
{"accessKeyId":"ak_info","secretAccessKey":"sk_info"}
```

AI DataLake Flink Jar作业编辑界面设置作业参数。

类名（示例，请替换为用户Jar包中实际的主类全限定名）：
1

com.example.dew.DataGen2FileSystemSink

参数：

     --output.path obs://USER_BUCKET_NAME/output/DataGen2FileSystemSink

优化参数：

     flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.access.key=USER_AK_CSMS_KEY
flink.hadoop.fs.obs.bucket.USER_BUCKET_NAME.dew.secret.key=USER_SK_CSMS_KEY
flink.hadoop.fs.obs.security.provider=com.aidatalake.flink.auth.provider.DewObsCredentialProvider
flink.hadoop.fs.dew.csms.secretName=obsAksK
flink.hadoop.fs.dew.csms.version=v1
flink.hadoop.fs.dew.csms.cache.time.second=3600
 
 
  

Flink Jar作业示例。

环境准备

已安装和配置IntelliJ IDEA等开发工具以及安装JDK和Maven。

pom文件配置中依赖包：

      <properties>
        <flink.version>1.20.0</flink.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- fastjson -->
        <dependency>
            <artifactId>fastjson</artifactId>
            <version>2.0.15</version>
        </dependency>
    </dependencies>
 
 
  

示例代码：

     package com.example.dew;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Random;

public class DataGen2FileSystemSink {
    private static final Logger LOG = LoggerFactory.getLogger(DataGen2FileSystemSink.class);

    public static void main(String[] args) {
        ParameterTool params = ParameterTool.fromArgs(args);
        LOG.info("Params: " + params.toString());
        try {
            StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();

            // checkpoint存储目录由平台自动注入（execution.checkpointing.dir为系统保留参数），无需自定义
            streamEnv.setStateBackend(new EmbeddedRocksDBStateBackend());
            streamEnv.getCheckpointConfig().setExternalizedCheckpointCleanup(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
            streamEnv.enableCheckpointing(30 * 1000);

            DataStream<String> stream = streamEnv.addSource(new DataGen())
                .setParallelism(1)
                .disableChaining();

            String outputPath = params.get("output.path", "obs://bucket/outputPath/jobId_jobName");

            // Sink OBS
            final StreamingFileSink<String> sinkForRow = StreamingFileSink
                .forRowFormat(new Path(outputPath), new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(OnCheckpointRollingPolicy.build())
                .build();

            stream.addSink(sinkForRow);

            streamEnv.execute("sinkForRow");
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }
    }
}

class DataGen implements ParallelSourceFunction<String> {

    private boolean isRunning = true;

    private Random random = new Random();

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (isRunning) {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("id", random.nextLong());
            jsonObject.put("name", "Molly" + random.nextInt());
            jsonObject.put("address", "hangzhou" + random.nextInt());
            jsonObject.put("birthday", System.currentTimeMillis());
            jsonObject.put("city", "hangzhou" + random.nextInt());
            jsonObject.put("number", random.nextInt());
            ctx.collect(jsonObject.toJSONString());
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
 
 
  

父主题：创建AI DataLake Flink引擎端点

上一篇：Flink SQL使用DEW管理访问凭据

下一篇：获取Flink作业委托临时凭证用于访问其他云服务