使用 Python 提交数据探查查询

步骤一：安装依赖

在终端执行以下命令安装 DLF Python SDK 和 PyArrow：

pip install alibabacloud-dlfnext20250310==3.8.0 pyarrow

步骤二：配置环境变量

设置运行所需的访问凭证和地域信息：

export ALIBABA_CLOUD_ACCESS_KEY_ID='<AccessKey ID>'
export ALIBABA_CLOUD_ACCESS_KEY_SECRET='<AccessKey Secret>'
export DLF_REGION_ID='cn-hangzhou'
export DLF_DEFAULT_CATALOG='dlf_samples'

# 可选：公网访问 Endpoint（SDK 通常根据地域自动解析，无需手动配置）
# export DLF_ENDPOINT='dlfnext.cn-hangzhou.aliyuncs.com'

步骤三：运行查询并读取结果

将以下代码保存为 data_exploration_query_demo.py：

import json
import os
import time
from urllib.request import urlopen

import pyarrow as pa
from alibabacloud_dlfnext20250310.client import Client
from alibabacloud_dlfnext20250310.models import GetQueryRequest, SubmitQueryRequest
from alibabacloud_tea_openapi.models import Config

TERMINAL_STATUSES = {"COMPLETED", "FAILED", "TIMEOUT", "CANCELLED"}


def required_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing environment variable: {name}")
    return value


def create_client() -> Client:
    return Client(Config(
        access_key_id=required_env("ALIBABA_CLOUD_ACCESS_KEY_ID"),
        access_key_secret=required_env("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
        security_token=os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN"),
        region_id=required_env("DLF_REGION_ID"),
        endpoint=os.getenv("DLF_ENDPOINT"),
    ))


def wait_for_result(client: Client, query_id: str, timeout_seconds: int = 300):
    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        body = client.get_query(query_id, GetQueryRequest()).body
        if body and body.status in TERMINAL_STATUSES:
            return body
        time.sleep(0.5)
    raise TimeoutError(f"Query {query_id} did not finish within {timeout_seconds}s")


def read_arrow_rows(download_url: str):
    with urlopen(download_url, timeout=30) as response:
        payload = response.read()
    table = pa.ipc.open_file(pa.BufferReader(payload)).read_all()
    return table.column_names, table.to_pylist()


def main():
    client = create_client()
    request = SubmitQueryRequest(
        default_catalog=required_env("DLF_DEFAULT_CATALOG"),
        sql="SELECT * FROM dlf_samples.search_samples.berkeley_deepdrive_100k_images LIMIT 10",
        tier="standard",
        limit=1000,
    )

    query_id = client.submit_query(request).body.query_id
    result = wait_for_result(client, query_id)

    if result.status != "COMPLETED":
        errors = [
            f"{item.error_code}: {item.error}"

            for item in (result.results or [])

            if item.error_code or item.error
        ]
        raise RuntimeError(
            f"Query {query_id} ended as {result.status}: {'; '.join(errors)}"
        )

    statement = result.results[0]
    print(f"query_id={query_id}, row_count={statement.row_count}")
    if not statement.download_url:
        print("Query completed with no result rows.")
        return

    columns, rows = read_arrow_rows(statement.download_url)
    print("columns:", columns)
    print(json.dumps(rows, ensure_ascii=False, default=str, indent=2))


if __name__ == "__main__":
    main()

在终端中执行以下命令运行脚本：

python data_exploration_query_demo.py