文档首页/ 智能数据湖 AI DataLake/ 用户指南/ 创建AI DataLake Flink引擎端点/ 获取Flink作业委托临时凭证用于访问其他云服务
更新时间：2026-10-21 GMT+08:00
查看PDF
分享

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

什么是临时凭证？

临时安全凭证是具备临时访问权限的身份凭证，包括临时AK/SK和SecurityToken，临时AK/SK和SecurityToken必须同时使用。

了解临时安全凭证。

功能描述

AI DataLake提供了一个通用接口，可用于获取作业所使用委托的临时凭证。作业使用的委托在创建端点时配置，详见约束限制。该接口将获取到的作业委托的临时凭证封装到com.huaweicloud.sdk.core.auth.BasicCredentials类中。

  • 获取到的委托的临时认证封装到com.huaweicloud.sdk.core.auth.ICredentialProvider接口的getCredentials()返回值中。
  • 返回类型为com.huaweicloud.sdk.core.auth.BasicCredentials。
  • 仅支持获取AK、SK、SecurityToken。
  • ICredentialProvider的实现已内置于AI DataLake Flink引擎中，并通过SPI机制注册，用户作业中无需自行实现该接口，直接使用"示例代码"章节提供的工具类加载获取即可。
  • 获取到AK、SK、SecurityToken后，请参考如何使用凭据管理服务替换硬编码的数据库账号密码查询凭据。

    使用临时AK、SK访问华为云凭据管理服务（CSMS），需在参考文档中getCsmsClient()方法的配置基础上，添加临时凭证必需的SecurityToken配置：

    private static CsmsClient getCsmsClient() {
        BasicCredentials auth = new BasicCredentials()
                .withAk(ACCESS_KEY)
                .withSk(SECRET_ACCESS_KEY)
                .withSecurityToken(securityToken)
                .withProjectId(PROJECT_ID);

约束限制

  • 仅支持Flink1.20版本使用委托授权访问临时凭证。AI DataLake Flink引擎当前为Flink1.20版本，无需额外选择作业引擎版本。
  • 已在创建端点时配置允许AI DataLake访问DEW的委托信息（端点凭证配置中的委托名称）。作业获取临时凭证时使用该委托：
    • 提交作业时未指定作业级委托，则默认使用端点配置的委托。
    • 提交作业时，也可以通过作业请求参数job_agency指定作业级委托。指定的委托必须已存在，且不能使用系统管理委托。
    • 委托的临时凭证由平台自动注入作业运行环境，作业代码中通过工具类接口获取即可，无需在作业配置中额外指定委托参数。

      委托权限示例请参考创建自定义委托和常见场景的委托权限策略。

  • AI DataLake Flink引擎基础镜像内已内置3.1.177版本的huaweicloud-sdk-core（版本随引擎发布变化，以引擎实际内置版本为准），无需重复安装。请在pom文件配置中将该依赖声明为provided，避免打包进用户Jar。

准备环境

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

Maven工程的pom.xml文件配置请参考JAVA样例代码中“pom文件配置”说明。

pom文件配置中依赖包

1
2
3
4
5
6
 
        <dependency>
            <groupId>com.huaweicloud.sdk</groupId>
            <artifactId>huaweicloud-sdk-core</artifactId>
            <version>3.1.177</version>
            <scope>provided</scope>
        </dependency>

如需使用临时凭证访问华为云凭据管理服务（DEW CSMS）查询凭据，还需增加huaweicloud-sdk-csms依赖，并排除其传递依赖的huaweicloud-sdk-core，保持SDK core由引擎提供，避免用户Jar中出现两份SDK core：

        <dependency>
            <groupId>com.huaweicloud.sdk</groupId>
            <artifactId>huaweicloud-sdk-csms</artifactId>
            <version>3.1.177</version>
            <exclusions>
                <exclusion>
                    <groupId>com.huaweicloud.sdk</groupId>
                    <artifactId>huaweicloud-sdk-core</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

示例代码

本章节JAVA样例代码演示如何获取BasicCredentials，以及获取作业委托的AK、SK、SecurityToken。

示例代码中的CSMS_ENDPOINT、PROJECT_ID、SECRET_NAME需按实际情况替换：

  • CSMS_ENDPOINT：华为云CSMS服务访问终端地址，与目标凭据所在区域对应，格式为https://kms.<region>.myhuaweicloud.com，可从DEW API参考“使用前必读”的“终端节点”章节查询。
  • PROJECT_ID：华为云局点项目ID，取CSMS_ENDPOINT对应区域的项目ID，在控制台右上角“用户名 > 我的凭证 > 项目列表”中查看。
  • SECRET_NAME：DEW凭据管理服务中目标凭据的名称。
  • Flink UDF 获取作业委托凭证 
    package com.aidatalake.flink.demo;

import static com.aidatalake.flink.demo.utils.AIDataLakeJobAgencyCredentialUtils.getICredentialProvider;

import com.huaweicloud.sdk.core.auth.BasicCredentials;
import com.huaweicloud.sdk.core.auth.ICredentialProvider;
import com.huaweicloud.sdk.csms.v1.CsmsClient;
import com.huaweicloud.sdk.csms.v1.model.Secret;
import com.huaweicloud.sdk.csms.v1.model.ShowSecretRequest;
import com.huaweicloud.sdk.csms.v1.model.ShowSecretResponse;

import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.ScalarFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class GetUserAgencyCredentialUDF extends ScalarFunction {
    private static final Logger LOG = LoggerFactory.getLogger(GetUserAgencyCredentialUDF.class);

    // 按实际情况替换：DEW CSMS接入地址与项目ID
    private static final String CSMS_ENDPOINT = "https://kms.<region>.myhuaweicloud.com";
    private static final String PROJECT_ID = "your-project-id";

    ICredentialProvider credentialProvider;
    CsmsClient csmsClient;

    @Override
    public void open(FunctionContext context) throws Exception {
        // 通过SPI获取作业委托临时凭证，无需在作业配置中额外指定委托参数
        credentialProvider = getICredentialProvider();

        // 临时AK/SK和SecurityToken已封装在BasicCredentials中（SecurityToken为临时凭证必需项），指定项目ID后构建CSMS客户端
        BasicCredentials basicCredentials = (BasicCredentials) credentialProvider.getCredentials();
        basicCredentials = basicCredentials.withProjectId(PROJECT_ID);
        csmsClient = CsmsClient.newBuilder()
                .withCredential(basicCredentials)
                .withEndpoints(List.of(CSMS_ENDPOINT))
                .build();
    }

    public String eval(String secretName) {
        ShowSecretResponse response = csmsClient.showSecret(
                new ShowSecretRequest().withSecretName(secretName));
        Secret secret = response.getSecret();
        if (secret == null) {
            throw new RuntimeException("DEW secret not found: " + secretName);
        }
        LOG.info("Accessed DEW CSMS with agency temp credentials, secret name={}", secret.getName());
        // 返回凭据元数据证明使用作业委托的临时凭证访问DEW成功；实际业务中可在此获取凭据值用于后续访问其他云服务
        return secret.getName();
    }
}

    将上述UDF类及工具类打包为Jar并上传至OBS后，即可在Flink SQL作业中通过CREATE TEMPORARY FUNCTION语句注册该UDF，并在SELECT中使用作业委托的临时凭证查询DEW凭据：

    CREATE TEMPORARY FUNCTION get_dew_credential AS 'com.aidatalake.flink.demo.GetUserAgencyCredentialUDF'
  USING JAR 'obs://<bucket>/<path>/agency-credential-udf.jar';
SELECT get_dew_credential('secret-name');

    USING JAR指定的Jar在SQL编译时自动下载注册，无需在作业配置中额外添加依赖；作业运行成功即表示已使用作业委托的临时凭证成功访问DEW。

  • Flink Jar作业获取作业委托凭证 
    package com.aidatalake.flink.demo;

import static com.aidatalake.flink.demo.utils.AIDataLakeJobAgencyCredentialUtils.getICredentialProvider;

import com.huaweicloud.sdk.core.auth.BasicCredentials;
import com.huaweicloud.sdk.core.auth.ICredentialProvider;
import com.huaweicloud.sdk.csms.v1.CsmsClient;
import com.huaweicloud.sdk.csms.v1.model.Secret;
import com.huaweicloud.sdk.csms.v1.model.ShowSecretRequest;
import com.huaweicloud.sdk.csms.v1.model.ShowSecretResponse;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;

public class GetUserCredentialsFlinkStream {
    // 按实际情况替换：DEW CSMS接入地址、项目ID与凭据名
    private static final String CSMS_ENDPOINT = "https://kms.<region>.myhuaweicloud.com";
    private static final String PROJECT_ID = "your-project-id";
    private static final String SECRET_NAME = "your-secret-name";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();

        streamEnv.fromElements(1)
                .map(i -> {
                    ICredentialProvider credentialProvider = getICredentialProvider();
                    // 临时AK/SK和SecurityToken已封装在BasicCredentials中（SecurityToken为临时凭证必需项），指定项目ID后构建CSMS客户端
                    BasicCredentials basicCredentials = (BasicCredentials) credentialProvider.getCredentials();
                    basicCredentials = basicCredentials.withProjectId(PROJECT_ID);

                    CsmsClient client = CsmsClient.newBuilder()
                            .withCredential(basicCredentials)
                            .withEndpoints(List.of(CSMS_ENDPOINT))
                            .build();

                    // 使用作业委托的临时凭证查询DEW凭据
                    ShowSecretResponse response = client.showSecret(
                            new ShowSecretRequest().withSecretName(SECRET_NAME));
                    Secret secret = response.getSecret();
                    if (secret == null) {
                        throw new RuntimeException("DEW secret not found: " + SECRET_NAME);
                    }
                    return "Accessed DEW CSMS with job agency credential, secret name=" + secret.getName()
                            + ", id=" + secret.getId();
                })
                .print();

        streamEnv.execute("GetUserCredentialsFlinkStream");
    }
}
  • 获取作业委托的工具类 
     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
    		 
    package com.aidatalake.flink.demo.utils;

import com.huaweicloud.sdk.core.auth.ICredentialProvider;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;
import java.util.ServiceLoader;

public class AIDataLakeJobAgencyCredentialUtils {

    public static ICredentialProvider getICredentialProvider() {
        List<ICredentialProvider> credentialProviders = new ArrayList<>();
        ServiceLoader.load(ICredentialProvider.class, StreamExecutionEnvironment.class.getClassLoader())
            .iterator()
            .forEachRemaining(credentialProviders::add);

        if (credentialProviders.size() != 1) {
            throw new RuntimeException("Failed to obtain temporary user credential");
        }
        return credentialProviders.get(0);
    }
}
父主题： 创建AI DataLake Flink引擎端点

相关文档