1. 在开发机中运行以下Python脚本，创建一个新的 Lance 数据集，并将视频数据转为Lance 数据集的 Lance 格式写入新的 Lance 数据集，核心示例代码如下。

   # 请在 LAS 开发机中执行以下代码
   import os
   
   """
   转为 Lance 数据集
   """
   import daft
   from daft.io import CreateLasDatasetOptions, IOConfig, LanceWriteOptions
   from daft.las.io import TOSConfig
   
   # 您可以自定义您新数据集的路径
   dataset_name = "las_dataset_video"
   lance_tos_dir = f"tos://las-doctest/video2lance/{dataset_name}.lance"
   LANCE_DATASET_FORMAT = "lance"
   
   # 创建相关的配置
   io_config = IOConfig(s3=TOSConfig.from_env().to_s3_config())
   write_options = LanceWriteOptions(io_config=io_config, uri=lance_tos_dir)
   create_ds_options = CreateLasDatasetOptions(
       nick_name="daft_test_lance_write",
       privacy="public",
       description="This is my dataset",
   )
   
   # 读取原始数据
   df = daft.read_las_dataset(name=dataset_name)
   
   # 在这里进行创建
   new_dataset_name = dataset_name + "_lance"
   df.write_las_dataset(
       name=new_dataset_name,
       format=LANCE_DATASET_FORMAT,
       write_options=write_options,
       create_ds_options=create_ds_options,
   )
   
   # 读取转换后的新数据集
   df = daft.read_las_dataset(name=new_dataset_name)
   print("\n\nnew dataset:")
   df.show()
   

您需要修改以下参数。

• dataset_name：设置为视频数据集名称，本示例为las_dataset_video
• lance_tos_dir：设置上述步骤中准备好的转换后的新Lance数据集的TOS路径。

其他参数可保持示例值即可。

new dataset:
╭────────────────────────────────┬───────────┬──────────╮
│ video                          ┆ size      ┆ num_rows │
│ ---                            ┆ ---       ┆ ---      │
│ String                         ┆ Int64     ┆ Int64    │
╞════════════════════════════════╪═══════════╪══════════╡
│ s3://las-doctest/video/LAS AI… ┆ 225602609 ┆ None     │
╰────────────────────────────────┴───────────┴──────────╯

转换后的 Lance 数据集包含“video”、“size”、“num_rows”字段。其中：

• video字段：为当前 Lance 数据集中每个视频数据所在的 TOS 路径（s3格式）。
• size字段：为每个视频的大小。

回显中如果出现ERROR提示，但是正常返回了数据集数据，可忽略ERROR提示。

# 地域设置，本示例使用华北2-北京地域
export REGION="cn-beijing"

# 鉴权AK相关环境变量
# 访问TOS、LAS的认证sk、ak
export LAS_TOS_SECRET_KEY="<your_sk>"
export LAS_TOS_ACCESS_KEY="<your_ak>"

# TOS 访问相关环境变量
# TOS的访问端点
export LAS_TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"
# TOS端点（用于TOSConfig）
export TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"

# DAFT执行器类型，设置为本地模式
export DAFT_RUNNER="native" 

本步骤新增了：

• DAFT_RUNNER：设置处理视频数据算子的Daft执行器为本地模式。

在开发机中执行以下脚本。

# 从未来版本导入特性，确保兼容性
from __future__ import annotations

# 导入必要的库
import base64  # 用于Base64编码和解码
import hashlib  # 用于生成SHA256哈希
import os  # 用于操作系统交互，如环境变量获取

import pyarrow as pa  # 用于处理列式数据

import daft  # 导入daft数据处理框架
from daft import DataType, col  # 导入数据类型和列操作工具
from daft.io import (
    CreateLasDatasetOptions,  # 用于创建LAS数据集的配置选项
    IOConfig,  # IO配置类
    LanceWriteOptions,  # Lance格式写入选项
)
from daft.io.object_store_options import io_config_to_storage_options  # 转换IO配置为存储选项
from daft.daft import IOConfig  # 再次导入IOConfig（可能是为了确保正确性）
from daft.las.functions.udf import las_udf  # 导入LAS用户定义函数装饰器
from daft.las.functions.video import VideoKeyframes  # 导入视频关键帧处理函数
from daft.las.io.tos import TOSConfig  # 导入TOS存储配置


# 设置视频源文件所在的TOS路径
TOS_TEST_DIR = "tos://las-doctest/video/LAS AI Product Introduction.mp4"
# 设置视频抽帧后的结果图片存储的TOS路径
TOS_TEST_OUTPUT_DIR = "tos://las-doctest/video/result/"

# 将视频抽帧结果创建为一个新的Lance数据集，以下参数用于配置新的Lance数据集参数
# 自定义新的Lance数据集名称
DATASET_NAME = "las_dataset_video_lance_kf"
# 设置新的Lance数据集的TOS路径
TOS_TEST_LANCE_DIR = "tos://las-doctest/video2lance/las_dataset_video_kf.lance"
# 抽帧后，写入Lance数据集中，自定义的视频路径列名
VIDEO_COLUMN = os.getenv("VIDEO_COLUMN", "video")
# 抽帧后，写入Lance数据集中，自定义的视频关键帧列名
VIDEO_KEYFRAMES_COLUMN = os.getenv("VIDEO_KEYFRAMES_COLUMN", "video_keyframes")
# 抽帧后，写入Lance数据集中，自定义的视频关键帧所在TOS路径列名
VIDEO_KEYFRAMES_TOS_COLUMN = os.getenv("VIDEO_KEYFRAMES_TOS_COLUMN", "video_keyframes_tos")


# 如果DAFT执行器是ray，则初始化ray
if os.environ['DAFT_RUNNER'] == 'ray':
    import ray  # 导入ray分布式计算框架


    ray.init(address='auto')  # 自动连接到ray集群


# 定义DAFT UDF：将Base64字符串转换为二进制数据
# ✅ 新写法：@daft.func 替代 @daft.udf（无状态函数）
@daft.func
def base64_to_binary(base64_str: DataType.string()) -> DataType.binary():
    import base64 as b64
    return b64.b64decode(base64_str)


# 定义DAFT UDF：生成SHA256哈希值
@daft.func
def generate_sha256_hash(s: DataType.string()) -> DataType.string():
    import hashlib
    return hashlib.sha256(str(s).encode()).hexdigest()


# 主函数入口
if __name__ == "__main__":
    # 从环境变量配置创建IO配置（适配TOS存储）
    io_config = IOConfig(s3=TOSConfig.from_env().to_s3_config())


    # 创建包含视频路径的样本数据
    samples = {VIDEO_COLUMN: [f"{TOS_TEST_DIR}"]}
    # 从字典创建daft数据集
    ds = daft.from_pydict(samples)
    
    # 创建视频关键帧提取器UDF
    extractor = las_udf(
        VideoKeyframes,  # 使用视频关键帧提取函数
        construct_args={
            "method": "I_frame",  # 提取方法：I帧
            "keyframes_cnt": 50,  # 提取的关键帧数量
            "output_tos_dir": f"{TOS_TEST_OUTPUT_DIR}",  # 输出到TOS的目录
        },
    )
    
    # 应用提取器，添加关键帧列到数据集
    ds = ds.with_column(VIDEO_KEYFRAMES_COLUMN, extractor(col(VIDEO_COLUMN)))


    # 从关键帧结果中提取TOS路径列
    ds = ds.with_column(VIDEO_KEYFRAMES_TOS_COLUMN, col(VIDEO_KEYFRAMES_COLUMN)["tos_paths"])


    # 从关键帧结果中提取base64编码的图像数据
    ds = ds.with_column(VIDEO_KEYFRAMES_COLUMN, col(VIDEO_KEYFRAMES_COLUMN)["base64"])
    # 对关键帧数据和对应的TOS路径进行展开（一行变多行）
    ds = ds.explode(col(VIDEO_KEYFRAMES_COLUMN), col(VIDEO_KEYFRAMES_TOS_COLUMN))
    # 将base64编码的图像数据转换为二进制
    ds = ds.with_column(VIDEO_KEYFRAMES_COLUMN, base64_to_binary(col(VIDEO_KEYFRAMES_COLUMN)))
    # 基于TOS路径生成数据项ID（SHA256哈希）
    ds = ds.with_column("__data_item_id", generate_sha256_hash(col(VIDEO_KEYFRAMES_TOS_COLUMN)))


    # 设置数据存储格式为lance
    format = "lance"
    # 配置Lance写入选项
    write_options = LanceWriteOptions(uri=TOS_TEST_LANCE_DIR, io_config=io_config)
    # 打印IO配置（用于调试）
    print(io_config)
    
    # 配置创建LAS数据集的选项
    create_ds_options = CreateLasDatasetOptions(
        nick_name=DATASET_NAME,  # 数据集昵称
        privacy="public",  # 隐私设置：公开
        description="This is test dataset"  # 数据集描述
    )
    
    # 将数据集写入LAS
    ds.write_las_dataset(
        name=DATASET_NAME,  # 数据集名称
        format=format,  # 存储格式
        write_options=write_options,  # 写入选项
        create_ds_options=create_ds_options,  # 创建数据集选项
    )
    
# 读取数据集
df = daft.read_las_dataset(name=DATASET_NAME)
df.show()    

您需要根据实际情况修改一些参数：

• TOS_TEST_DIR：设置为需要处理的视频文件所在的TOS路径。
• TOS_TEST_OUTPUT_DIR：设置为视频抽帧后，抽帧图片所在的TOS路径。
• DATASET_NAME：设置为新建的视频抽帧结果的 Lance 数据集名称。
• TOS_TEST_LANCE_DIR：新增的 Lance 数据集的 TOS 路径。
• VIDEO_COLUMN/VIDEO_KEYFRAMES_COLUMN/VIDEO_KEYFRAMES_TOS_COLUMN：视频抽帧处理完成后，写入 Lance 数据的原始视频TOS路径/视频关键帧/视频关键帧所在TOS路径的列名。

您还可按需修改调用视频关键帧抽取算子的算子相关参数：

• method：抽帧方法，本示例设置为I_frame。
• keyframes_cnt：提取的关键帧数量，本示例设置为 50，提取50 张关键帧图片。