用于批量导出指定数据集中的数据。
本功能包括两个接口：

• /task/create 创建导出任务
• /task/info 查询任务信息

创建导出任务(create)

/task/create 用于在指定的数据集 Collection 按条件批量导出任务。

请求接口

请求参数

{
    "user_id":[1,2,3]，
    "city":["beijing"]
}

响应消息

任务运行信息(info)

/task/info用于查询任务状态、任务进度等离线任务的详细信息。

请求接口

请求参数

响应消息

任务状态说明

错误码说明

1. 申请 tos 服务，并对向量库进行授权

申请开通 tos 服务 https://console.volcengine.com/tos/bucket/create
对向量库进行授权 https://console.volcengine.com/iam/service/attach_role/?ServiceName=ml_platform

2. 创建导出任务

import json, ijson
import logging
import os
import pandas as pd
import requests
from volcengine.auth.SignerV4 import SignerV4
from volcengine.Credentials import Credentials
from volcengine.base.Request import Request
from volcengine.viking_db import VikingDBService, Collection, common
import tos
import pyarrow.parquet as pq
import numpy as np
from pre import *

def prepare_request(method, path, ak, sk, params=None, data=None, doseq=0):
    request = Request()
    request.set_method(method)
    request.set_host("api-vikingdb.volces.com")
    request.set_path(path)
    mheaders = {
        'Accept': 'application/json',
        'Content-Type': 'application/json',
    }
    request.set_headers(mheaders)
    if params:
        request.set_query(params)
    if data:
        request.set_body(json.dumps(data))
    credentials = Credentials(ak, sk, 'air', 'cn-north-1')
    SignerV4.sign(request, credentials)
    return request

DOMAIN = "api-vikingdb.volces.com"
AK = "****"
SK = "****"
COLLECTION_NAME = "example"
BUCKET_NAME = "data-export"
TOS_DIR = "task_example"
task_type = "data_export"
file_type = "parquet"

task_param = {
    "collection_name": COLLECTION_NAME,
    "tos_path": "{}/{}".format(BUCKET_NAME, TOS_DIR),
    "filter" : {
        "user_id": [5,9],
        "city": ["beijing"]
    }
}

req_info = prepare_request(method = "POST", 
                    path = "/api/task/create",
                    ak = AK, 
                    sk = SK,
                    data ={"task_type":task_type, "task_params":task_param})
r = requests.request(method=req_info.method,
        url="http://{}{}".format(DOMAIN, req_info.path),
        headers=req_info.headers,
        data=req_info.body
    )
task_id = json.loads(r.text)['data']['task_id']

3. 扫描从 tos 中下载文件

def download(client, bucket_name, object_key, local_path):
    file_path = "{}/{}".format(local_path, object_key)
    try:
        client.get_object_to_file(bucket_name, object_key, file_path)
    except tos.exceptions.TosClientError as e:
        # 操作失败，捕获客户端异常，一般情况为非法请求参数或网络异常
        return 'fail with client error, message:{}, cause: {}'.format(e.message, e.cause)
    except tos.exceptions.TosServerError as e:
        return 'fail with server error : {}'.format(e)
    except Exception as e:
        return 'fail with unknown error: {}'.format(e)
    return ''


client = tos.TosClientV2(AK, SK, TOS_ENDPOINT, REGION)
download(client, BUCKET_NAME, "tos-file-path", LOCAL_FILE_PATH)

4. 解析parquet类型数据

def process_parquet(file_path):
    parquet_file = pq.ParquetFile(file)

    file_data_count = sum(parquet_file.metadata.row_group(i).num_rows for i in range(parquet_file.num_row_groups))
    schema = parquet_file.schema.to_arrow_schema()
    row_iterator = parquet_file.iter_batches(batch_size=100)
    # 迭代读取数据
    for batch in row_iterator:
        df = batch.to_pandas()
        for index, row in df.iterrows():
            """
            do what you want
            """
            print(row)
    return ''

process_parquet("1.parquet")