环境变量设置

env_vars允许在UDF调用时动态注入环境变量到Actor进程，无需为不同环境变量配置创建冗余UDF定义。

基本用法

# Dataset API — 传入 dict
ds = ds.map(
    fn="my_schema.data_loader",
    on=["url_col"],
    env_vars={"DATA_SOURCE": "obs://bucket/path", "REGION": "cn-north-4"},
)

# with_arguments — 传入 dict 或 JSON 字符串
result = my_udf(t.col_a).with_arguments(
    env_vars={"DATA_SOURCE": "obs://bucket/path"},
)

在UDF中通过os.environ访问：

def my_udf(url: str) -> str:
    import os
    data_source = os.environ.get("DATA_SOURCE", "")
    return load_from(url, base=data_source)

类型校验

Python侧在with_arguments() 中执行严格类型校验：

输入	行为
env_vars={"KEY": "VALUE"}	正常，key/value 均为 str
env_vars={1: "val"}	TypeError: keys must be str
env_vars={"key": 123}	TypeError: values must be str
env_vars={}	等同未设置，不生成 ENV_VARS 子句
env_vars='{"K":"V"}'	JSON 字符串直通
env_vars='not json'	TypeError: must be valid JSON
env_vars=None	默认值，不生成 ENV_VARS 子句

Key格式约束

env_vars的key必须符合POSIX规范^[a-zA-Z_][a-zA-Z0-9_]*$，最大长度1022字节。内核侧校验不合法时返回ERROR。

Key 示例	合法	说明
MY_KEY	是	标准POSIX key
_UNDERSCORE_KEY	是	以下划线开头
KEY123	是	包含数字但非首字符
1KEY	否	以数字开头
MY-KEY	否	包含连字符
MY.KEY	否	包含点号
MY=KEY	否	包含等号
my key	否	包含空格

安全注意事项

禁止在env_vars中传入AK/SK等敏感凭证

# 禁止 — 凭证明文经过 SQL 传输，有泄漏风险
my_udf(t.col).with_arguments(
    env_vars={"ACCESS_KEY_ID": "xxx", "SECRET_ACCESS_KEY": "yyy"}
)

# 禁止 — UDF_TEMP_AK/UDF_TEMP_SK是服务端凭证保留 key，用户设置会触发ERROR
my_udf(t.col).with_arguments(
    env_vars={"UDF_TEMP_AK": "xxx", "UDF_TEMP_SK": "yyy"}
)
# ERROR: env_vars key "UDF_TEMP_AK" conflicts with reserved server credentials, UDF 调用终止

使用内核IAM委托Token自动注入

内核通过IAM委托Token自动获取临时AK/SK，注入为UDF_TEMP_AK、UDF_TEMP_SK、UDF_TEMP_SECURITY_TOKEN环境变量。OBS Endpoint等非凭证配置通过用户env_vars传入。UDF内部直接通过os.environ读取，配合esdk-obs-python SDK使用：

class ObsDataReader:
    """Read data from OBS.

    Credentials (UDF_TEMP_AK/SK/TOKEN) are automatically injected by
    the kernel via IAM delegation token — no client-side setup needed.
    UDF_OBS_ENDPOINT must be provided by the user via env_vars.
    """

    def __init__(self, bucket: str):
        self.bucket = bucket

    def _get_obs_config(self) -> tuple:
        """Read OBS config from env vars."""
        import os

        # Credentials — auto-injected by kernel, do NOT set these yourself
        ak = os.environ.get("UDF_TEMP_AK", "")
        sk = os.environ.get("UDF_TEMP_SK", "")
        token = os.environ.get("UDF_TEMP_SECURITY_TOKEN", "")

        # Endpoint — user-provided via env_vars
        endpoint = os.environ.get("UDF_OBS_ENDPOINT", "")

        if not ak or not sk:
            raise RuntimeError(
                "OBS credentials not available. "
                "Ensure IAM delegation token is configured on the server."
            )
        if not endpoint:
            raise RuntimeError(
                "OBS endpoint not configured. "
                "Pass it via env_vars: with_arguments(env_vars={'UDF_OBS_ENDPOINT': '...'})"
            )

        return ak, sk, token, endpoint

    def __call__(self, object_key: str) -> str:
        """Read an object from OBS and return its content as string."""
        from obs import ObsClient

        ak, sk, token, endpoint = self._get_obs_config()

        obs_client = ObsClient(
            access_key_id=ak,
            secret_access_key=sk,
            security_token=token,
            server=endpoint,
        )

        try:
            resp = obs_client.getObject(
                self.bucket, object_key, loadStreamInMemory=True
            )
            if resp.status >= 300:
                raise RuntimeError(
                    f"OBS read failed: status={resp.status}, reason={resp.reason}"
                )
            return resp.body.buffer.decode("utf-8")
        finally:
            obs_client.close()

使用方式：

from aura_frame.multimodal import ai_lake

conn = ai_lake.connect(endpoint='*', ak='*', sk='*')

# Register the UDFs
conn.create_scalar_function(ObsDataReader, name="obs_reader", database="my_schema")

# --- ObsDataReader: read an object from OBS ---
# bucket via constructor param, endpoint via env_vars, credentials auto-injected
t = conn.from_pandas(pandas.DataFrame({"object_key": [""]}))._table
result = conn.get_function("obs_reader", database="my_schema")(
    t.object_key
).with_arguments(
    bucket="my-data-bucket",
    env_vars={"UDF_OBS_ENDPOINT": "obs.cn-north-4.myhuaweicloud.com"},
)

# Or via Dataset API
ds = conn.from_pandas(pandas.DataFrame({"object_key": [""]}))
ds = ds.map(
    fn="my_schema.obs_reader",
    on=["object_key"],
    bucket="my-data-bucket",
    as_col="content",
    env_vars={"UDF_OBS_ENDPOINT": "obs.cn-north-4.myhuaweicloud.com"},
)

服务端凭证保留key（用户env_vars中设置会触发ERROR）：

保留 key	用途
UDF_TEMP_AK	临时Access Key。
UDF_TEMP_SK	临时Secret Key。
UDF_TEMP_SECURITY_TOKEN	临时Security Token。
SHARED_WORK_DIR	会话级共享存储的路径，同一个session内的所有UDF实例都可以访问该路径，进行读写。