前期准备 #

该功能需安装火山引擎提供的Ray 2.46.0 及以上版本。

样例操作 #

写入数据 #

新版本写入

新版本中write_lance 方法已集成至 ray.data.Dataset 对象，成为其内置方法。

• 旧实现：需从 lance.ray.sink 模块导入 LanceDatasink 类，并手动实例化后使用；
• 新实现：无需额外导入或实例化操作，直接通过 ray.data.Dataset 对象调用即可。

import ray

# 初始化 Ray
ray.init()

# 创建 Ray Dataset
ds = ray.data.range(10).map(
    lambda x: {"id": x["id"], "str": f"str-{x['id']}"}
)

# 使用 write_lance 方法将数据写入 Lance 格式
ds.write_lance(
    path="s3://{PATH}",
    storage_options={
        "access_key_id": "{用户实际的AK}",
        "secret_access_key": "{用户实际的SK}",
        "aws_endpoint": "https://{bucket name}.{S3 Endpoint}.ivolces.com",
        "virtual_hosted_style_request": "true"
    },
    mode='create',
    max_rows_per_file=8192
)

print("数据已成功写入 Lance 格式。")

Ray的Lance写入, 已经封装为了LanceDatasink, 可以通过构建LanceDatasink对象, 调度write_datasink完成写入
write_lance参数

旧版本写入

import ray
from lance.ray.sink import LanceDatasink

ray.init()

sink = LanceDatasink(
       uri="s3://{PATH}",
       storage_options={
         "access_key_id": "{用户实际的AK}",
         "secret_access_key": "{用户实际的SK}",
         "aws_endpoint": "https://{bucket name}.{S3 Endpoint}.ivolces.com",
         "virtual_hosted_style_request": "true"
       }
)
ray.data.range(10).map(
    lambda x: {"id": x["id"], "str": f"str-{x['id']}"}
).write_datasink(sink)

使用Ray进行Lance数据写入，已经封装为LanceDatasink，可以通过构建LanceDatasink对象，调度write_datasink完成写入。
LanceDatasink参数说明

读取数据 #

import ray
ds = ray.data.read_lance( 
    uri="s3://hzw/tpcds/backup/store_sales_10g.lance",
    storage_options={
             "access_key_id": "{用户实际的AK}",
             "secret_access_key": "{用户实际的SK}",
             "aws_region": "shanghai",
             "aws_endpoint": "https://{bucket name}.{S3 Endpoint}.ivolces.com",
             "virtual_hosted_style_request": "true"
        },
    concurrency=1
)
ds.take(10)

参数说明

Parquet 格式转 Lance 样例 #

使用前需确认已安装 pyarrow 和 s3fs，便于使用 ray 读取 s3。如不确定，请参考 Ray 的使用文档。
以华北2为例：

• 如在火山引擎的ECS上运行，Endpoint的地址请使用： https://bucket1.tos-s3-cn-beijing.ivolces.com。
• 如在本地的机器上运行，使用公网地址： https://bucket1.tos-s3-cn-beijing.volces.com。

import ray
import pyarrow.fs as fs
# 初始化 Ray
ray.init()
# 1. 配置 PyArrow FileSystem 用于读取数据
#    注意：这里的 endpoint 和配置用于读取操作
read_fs = fs.S3FileSystem(
    access_key='{用户实际的AK}',
    secret_key='{用户实际的SK}',
    region='cn-beijing',
    endpoint_override='https://{S3 Endpoint}.volces.com',
    force_virtual_addressing=True
)
# 2. 从 S3 读取 Parquet 数据
ds = ray.data.read_parquet("s3://bucket1/{parquet_PATH}/", filesystem=read_fs)

# 3. 使用 write_lance 直接写入数据
#    原 LanceDatasink 的参数被直接传入 write_lance 方法
ds.write_lance(
    path="s3://{PATH}",
    storage_options={
        "access_key_id": "{用户实际的AK}",
        "secret_access_key": "{用户实际的SK}",
        "aws_region": "cn-beijing",
        # 注意：这里的 endpoint 和配置用于写入操作
        "aws_endpoint": "https://{bucket name}.{S3 Endpoint}.volces.com",
        "virtual_hosted_style_request": "true"
    }, 
     max_concurrent_tasks=10  # 'concurrency' 参数现在由 'max_concurrent_tasks' 替代
)

print("数据已成功转换为 Lance 格式并写入目标位置！")