1. Run the following Python script on the development machine to create a new Lance dataset, convert the video data to the Lance format of the Lance dataset, and write it to the new Lance dataset. The core sample code is as follows.

   # Please execute the following code on the LAS development machine
   import os
   
   """
   Convert to Lance Dataset
   """
   import daft
   from daft.io import CreateLasDatasetOptions, IOConfig, LanceWriteOptions
   from daft.las.io import TOSConfig
   
   # You can customize the path for your new dataset
   dataset_name = "las_dataset_video"
   lance_tos_dir = f"tos://las-doctest/video2lance/{dataset_name}.lance"
   LANCE_DATASET_FORMAT = "lance"
   
   # Create the relevant configuration
   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",
   )
   
   # Read the raw data
   df = daft.read_las_dataset(name=dataset_name)
   
   # Create here
   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,
   )
   
   # Read the newly converted dataset
   df = daft.read_las_dataset(name=new_dataset_name)
   print("\n\nnew dataset:")
   df.show()
   

You need to modify the following parameters.

• dataset_name: Set to the video dataset name. In this example, it is las_dataset_video
• lance_tos_dir: Set the TOS path for the newly converted Lance dataset prepared in the previous steps.

Other parameters can be kept as the example values.

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

The converted Lance dataset contains the fields "video", "size", and "num_rows". Details:

• The video field: the TOS path (in s3 format) of each video data in the current Lance dataset.
• The size field: the size of each video.

If an ERROR message appears in the output but the dataset data is returned normally, the ERROR message can be ignored.

# Region settings. In this example, North China 2 - Beijing region is used
export REGION="cn-beijing"

# Authentication AK-related environment variables
# Authentication sk and ak for accessing TOS and LAS
export LAS_TOS_SECRET_KEY="<your_sk>"
export LAS_TOS_ACCESS_KEY="<your_ak>"

# TOS access-related environment variables
# TOS access endpoint
export LAS_TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"
# TOS endpoint (for TOSConfig)
export TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"

# DAFT executor type, set to local mode
export DAFT_RUNNER="native" 

This step adds:

• DAFT_RUNNER: Set the Daft executor for the video data processing operator to local mode.

Run the following script on the development machine.

# Import features from future versions to ensure compatibility
from __future__ import annotations

# Import necessary libraries
import base64  # For Base64 encoding and decoding
import hashlib  # For generating SHA256 hashes
import os  # For operating system interactions, such as obtaining environment variables

import pyarrow as pa  # For processing columnar data

import daft  # Import the daft data processing framework
from daft import DataType, col  # Import data types and column operation tools
from daft.io import (
    CreateLasDatasetOptions,  # Configuration options for creating LAS datasets
    IOConfig,  # IO configuration class
    LanceWriteOptions,  # Lance format write options
)
from daft.io.object_store_options import io_config_to_storage_options  # Convert IO configuration to storage options
from daft.daft import IOConfig  # Import IOConfig again (possibly to ensure correctness)
from daft.las.functions.udf import las_udf  # Import the LAS user-defined function decorator
from daft.las.functions.video import VideoKeyframes  # Import the video key frame processing function
from daft.las.io.tos import TOSConfig  # Import the TOS storage configuration


# Set the TOS path where the video source file is located
TOS_TEST_DIR = "tos://las-doctest/video/LAS AI Product Introduction.mp4"
# Set the TOS path for storing the resulting images after video frame extraction
TOS_TEST_OUTPUT_DIR = "tos://las-doctest/video/result/"

# Create a new Lance dataset from the video frame extraction results; the following parameters are used to configure the new Lance dataset
# Customize the name of the new Lance dataset
DATASET_NAME = "las_dataset_video_lance_kf"
# Set the TOS path for the new Lance dataset
TOS_TEST_LANCE_DIR = "tos://las-doctest/video2lance/las_dataset_video_kf.lance"
# After frame extraction, customize the video path column name when writing to the Lance dataset
VIDEO_COLUMN = os.getenv("VIDEO_COLUMN", "video")
# After frame extraction, customize the video key frame column name when writing to the Lance dataset
VIDEO_KEYFRAMES_COLUMN = os.getenv("VIDEO_KEYFRAMES_COLUMN", "video_keyframes")
# After frame extraction, customize the TOS path column name for the video key frames when writing to the Lance dataset
VIDEO_KEYFRAMES_TOS_COLUMN = os.getenv("VIDEO_KEYFRAMES_TOS_COLUMN", "video_keyframes_tos")


# If the DAFT executor is ray, initialize ray
if os.environ['DAFT_RUNNER'] == 'ray':
    import ray  # Import the ray distributed computing framework


    ray.init(address='auto')  # Automatically connect to the ray cluster


# Define a DAFT UDF: convert a Base64 string to binary data
# ✅ New syntax: use @daft.func instead of @daft.udf (stateless function)
@daft.func
def base64_to_binary(base64_str: DataType.string()) -> DataType.binary():
    import base64 as b64
    return b64.b64decode(base64_str)


# Define a DAFT UDF: generate a SHA256 hash value
@daft.func
def generate_sha256_hash(s: DataType.string()) -> DataType.string():
    import hashlib
    return hashlib.sha256(str(s).encode()).hexdigest()


# Main function entry point
if __name__ == "__main__":
    # Create IO configuration from environment variable settings (adapted for TOS storage)
    io_config = IOConfig(s3=TOSConfig.from_env().to_s3_config())


    # Create sample data containing video paths
    samples = {VIDEO_COLUMN: [f"{TOS_TEST_DIR}"]}
    # Create a daft dataset from a dictionary
    ds = daft.from_pydict(samples)
    
    # Create a video key frame extractor UDF
    extractor = las_udf(
        VideoKeyframes,  # Use the video key frame extraction function
        construct_args={
            "method": "I_frame",  # Extraction method: I-frame
            "keyframes_cnt": 50,  # Number of extracted key frames
            "output_tos_dir": f"{TOS_TEST_OUTPUT_DIR}",  # Output directory in TOS
        },
    )
    
    # Apply the extractor to add a key frame column to the dataset
    ds = ds.with_column(VIDEO_KEYFRAMES_COLUMN, extractor(col(VIDEO_COLUMN)))


    # Extract the TOS path column from the key frame results
    ds = ds.with_column(VIDEO_KEYFRAMES_TOS_COLUMN, col(VIDEO_KEYFRAMES_COLUMN)["tos_paths"])


    # Extract the base64-encoded image data from the key frame results
    ds = ds.with_column(VIDEO_KEYFRAMES_COLUMN, col(VIDEO_KEYFRAMES_COLUMN)["base64"])
    # Expand the key frame data and corresponding TOS paths (one row becomes multiple rows)
    ds = ds.explode(col(VIDEO_KEYFRAMES_COLUMN), col(VIDEO_KEYFRAMES_TOS_COLUMN))
    # Convert the base64-encoded image data to binary
    ds = ds.with_column(VIDEO_KEYFRAMES_COLUMN, base64_to_binary(col(VIDEO_KEYFRAMES_COLUMN)))
    # Generate a data item ID (SHA256 hash) based on the TOS path
    ds = ds.with_column("__data_item_id", generate_sha256_hash(col(VIDEO_KEYFRAMES_TOS_COLUMN)))


    # Set the data storage format to lance
    format = "lance"
    # Configure Lance write options
    write_options = LanceWriteOptions(uri=TOS_TEST_LANCE_DIR, io_config=io_config)
    # Print the IO configuration (for debugging)
    print(io_config)
    
    # Configure options for creating the LAS dataset
    create_ds_options = CreateLasDatasetOptions(
        nick_name=DATASET_NAME,  # Dataset nickname
        privacy="public",  # Privacy setting: public
        description="This is test dataset"  # Dataset description
    )
    
    # Write the dataset to LAS
    ds.write_las_dataset(
        name=DATASET_NAME,  # Dataset name
        format=format,  # Storage format
        write_options=write_options,  # Write options
        create_ds_options=create_ds_options,  # Create dataset options
    )
    
# Read the dataset
df = daft.read_las_dataset(name=DATASET_NAME)
df.show()    

You need to modify some parameters as needed:

• TOS_TEST_DIR: Set to the TOS path where the video files to be processed are located.
• TOS_TEST_OUTPUT_DIR: Set to the TOS path where the extracted keyframe images are located after video keyframe extraction.
• DATASET_NAME: Set to the name of the Lance dataset for the newly created video keyframe extraction results.
• TOS_TEST_LANCE_DIR: The TOS path of the newly added Lance dataset.
• VIDEO_COLUMN/VIDEO_KEYFRAMES_COLUMN/VIDEO_KEYFRAMES_TOS_COLUMN: After video keyframe extraction is completed, the column names in the Lance dataset for the original video TOS path, video keyframes, and the TOS path where the video keyframes are located.

You can also modify the relevant parameters of the video keyframe extraction operator as needed:

• method: Keyframe extraction method. In this example, it is set to I_frame.
• keyframes_cnt: Number of keyframes to extract. In this example, it is set to 50, extracting 50 keyframe images.