大数据开发治理平台 DataWorks：节点调度配置

方式一：读取环境变量（推荐 Shell, Python）

系统会自动设置以下三个环境变量，您可以直接在代码中读取。

• LINKED_ROLE_ACCESS_KEY_ID：临时访问密钥 ID。

• LINKED_ROLE_ACCESS_KEY_SECRET：临时访问密钥 Secret。

• LINKED_ROLE_SECURITY_TOKEN：临时安全令牌。

代码示例 (Python)：

import os
import oss2

# 1. 从环境变量获取临时凭证
access_key_id = os.environ.get('LINKED_ROLE_ACCESS_KEY_ID')
access_key_secret = os.environ.get('LINKED_ROLE_ACCESS_KEY_SECRET')
security_token = os.environ.get('LINKED_ROLE_SECURITY_TOKEN')

# 检查是否成功获取
if not all([access_key_id, access_key_secret, security_token]):
    raise Exception("Failed to get linked role credentials from environment variables.")

# 2. 使用临时凭证初始化 OSS 客户端
# 假设您已为角色授予了访问 'your-bucket-name' 的权限
auth = oss2.StsAuth(access_key_id, access_key_secret, security_token)
bucket = oss2.Bucket(auth, 'http://oss-<regionID>-internal.aliyuncs.com', 'your-bucket-name')

# 3. 使用客户端访问 OSS 资源
try:
    # 列出 bucket 中的文件
    for obj in oss2.ObjectIterator(bucket):
        print('object name: ' + obj.key)
    print("Successfully accessed OSS with linked role.")
except oss2.exceptions.OssError as e:
    print(f"Error accessing OSS: {e}")

代码示例 (Shell)：

#!/bin/bash
access_key_id=${LINKED_ROLE_ACCESS_KEY_ID}
access_key_secret=${LINKED_ROLE_ACCESS_KEY_SECRET}
security_token=${LINKED_ROLE_SECURITY_TOKEN}

# 访问OSS，需替换regionID、bucket_name和file_name为真实信息
echo "ID："$access_key_id
echo "token:"$security_token
ls -al /home/admin/usertools/tools/

# 示例为通过ossutil从OSS指定路径下载文件到本地test_dw.py，并输出文件内容。
/home/admin/usertools/tools/ossutil64 cp --access-key-id $access_key_id --access-key-secret $access_key_secret --sts-token $security_token --endpoint http://oss-<regionID>-internal.aliyuncs.com oss://<bucket_name>/<file_name> test_dw.py
echo "************************ 获取成功 ************************，打印结果"
cat test_dw.py

方式二：使用 Credentials Client (推荐 Python)

代码示例 (Python)：

from alibabacloud_credentials.client import Client as CredentialClient
import oss2

# 1. 使用 SDK 自动获取凭证
# 它会自动查找环境变量中的 LINKED_ROLE_* 等凭证信息
cred_client = CredentialClient()
credential = cred_client.get_credential()

access_key_id = credential.get_access_key_id()
access_key_secret = credential.get_access_key_secret()
security_token = credential.get_security_token()

if not all([access_key_id, access_key_secret, security_token]):
    raise Exception("Failed to get linked role credentials via SDK.")

# 2. 使用凭证初始化 OSS 客户端
auth = oss2.StsAuth(access_key_id, access_key_secret, security_token)
bucket = oss2.Bucket(auth, 'http://oss-cn-hangzhou.aliyuncs.com', 'your-bucket-name')

# 3. 访问 OSS
print("Listing objects in bucket...")
for obj in oss2.ObjectIterator(bucket):
    print(' - ' + obj.key)
print("Successfully accessed OSS with linked role via SDK.")