Quick start for multimodal data processing

Background information

Training an embodied AI model to understand kitchen activities typically requires thousands of hours of egocentric video recorded from head-mounted cameras. Decoding that footage yields hundreds of millions of frames, most of which are static shots, blurred frames, or empty frames that carry no useful signal. Filtering high-quality training samples out of this redundancy is usually the bottleneck.

Doing this the traditional way means stitching several systems together: FFmpeg for frame extraction and transcoding, custom scripts for image cleanup, and a separate model service for labeling. This topic uses the multimodal operator marketplace in EMR Serverless Spark instead, so frame extraction, cleanup, and labeling all happen in a single Daft DataFrame pipeline, with no model service to deploy.

The pipeline in this topic combines three kinds of capabilities:

• Daft native API: daft.read_video_frames() extracts video frames in parallel.

• Multimodal operator: ImageBlackBorderCrop detects and crops black borders.

• Expression-level API: ai_query() calls a multimodal LLM to generate the action description. The model is served by the built-in Model Manager of EMR Serverless Spark. For more information, see Model service.

Prerequisites

Make sure that all of the following requirements are met before you start:

Step 1: Upload a video to OSS

Prepare an egocentric video in MP4 format and upload it to an OSS bucket in the same region as your workspace. Also reserve an output directory in the bucket for the extracted and cropped frames. If you do not have suitable footage at hand, use the sample video P01_101_part012_299.40s_310.00s.mp4, a 10-second egocentric clip of kitchen activity.

For example:

• Input video: oss://<YOUR_BUCKET>/dataset/kitchen-demo.mp4

• Output directory: oss://<YOUR_BUCKET>/dataset/daft-output/

Step 2: Create a notebook session

Log on to the EMR Serverless Spark console, go to your workspace, and create and start a notebook session. Pay attention to the engine version:

• Set Engine Version to esr-4.9.0 (Spark 3.5.2, Scala 2.12, Ray 2.55.1, Daft 0.7.17). Multimodal operators ship with the engine image. If the version does not match, the operator is reported as not found.

• Give the session at least the resources listed in the prerequisites. Otherwise, frame extraction and model calls queue up.

For details about creating and running notebooks, see Get started with notebook development. For details about engine versions, see Manage runtime environments.

Step 3: Run the processing pipeline

Create a Python cell in the notebook, paste the following code, replace INPUT_VIDEO and OUTPUT_DIR with the paths you prepared in Step 1, and run the cell. Alternatively, download the complete notebook daft_demo.ipynb, import it into the notebook session, and run it directly instead of pasting the code cell by cell.

import base64
import os
from pathlib import PurePosixPath

import daft
from daft import DataType, col, lit
from daft.emr.functions import ImageBlackBorderCrop, ai_query, emr_udf
from pypaimon.daft import read_paimon, write_paimon

# ==========================================
# 1. Global configuration and prompt
# ==========================================
# Replace with your own OSS input video and output directory
INPUT_VIDEO = "oss://<YOUR_BUCKET>/dataset/kitchen-demo.mp4"
OUTPUT_DIR = "oss://<YOUR_BUCKET>/dataset/daft-output/"

PROMPT = """
Role: You are an expert in analyzing egocentric human videos. Identify every distinct action segment in which the person's arms interact with an object.
Output requirements: Return the action segment performed by both hands, in English, within 20 words. 
"""


# ==========================================
# 2. Custom UDF definitions
# ==========================================
@daft.func(return_dtype=DataType.string())
def image_name(path: str, frame_index: int) -> str:
    """Group output frames by video name so that frames from different videos stay separate"""
    return f"{PurePosixPath(path).stem}/frame_{frame_index:06d}"


@daft.func(return_dtype=DataType.binary())
def decode_image(value: str) -> bytes:
    """Convert the Base64 crop result into a Paimon BLOB (bytes)"""
    return base64.b64decode(value)


# ==========================================
# 3. Core pipeline (frame extraction -> border cropping -> LLM labeling)
# ==========================================
results = (
    # Step 1: Extract video frames
    daft.read_video_frames(
        path=INPUT_VIDEO,
        image_height=580,
        image_width=740,
        # start_time=600.0,  # Push down the time range, for example start reading at minute 10
        sample_interval_seconds=1,  # Sample one frame every 2 seconds
        decode_thread_type="AUTO",
        decode_thread_count=4,  # Multi-threaded video decoding
    )
    # Step 2: Crop black borders (calls an EMR UDF)
    .with_column(
        "crop",
        emr_udf(
            ImageBlackBorderCrop,
            construct_args={
                "image_src_type": "image_ndarray",
                "detect_algorithm": "auto",
                "crop_sides": ["top", "bottom"],  # Choose which borders to crop
                "output_dir": OUTPUT_DIR,
                "image_format": "jpg",
                "quality": 85,
            },
            concurrency=2,
            batch_size=32,
        )(
            image=col("data"),
            image_name=image_name(col("path"), col("frame_index")),
        ),
    )
    # Step 3: Extract the image URL and the Base64 data
    .with_column("image_url", col("crop").get("image_path"))
    .with_column("frame_blob", decode_image(col("crop").get("base64")))
    # Step 4: Let the multimodal LLM read the OSS image and generate a scene description
    .with_column(
        "scene_description",
        ai_query(lit(PROMPT), data=col("image_url"), data_type="uri", service_name="qwen3.7-plus").get("content"),
    )
    # Step 5: Select and rename columns (keep only the fields to write to Paimon)
    .select(
        col("path").alias("video_path"),
        col("frame_index"),
        col("frame_time"),
        col("image_url"),
        col("frame_blob"),
        col("scene_description"),
        lit("2026-08-01").alias("dt"),
    )
)

# ==========================================
# 4. Trigger execution and print the result
# ==========================================
results.collect()
print(results)

The pipeline consists of four stages:

The code also defines two custom UDFs. image_name() names the output frames after the source video so that frames from different videos do not end up in the same directory. decode_image() converts the Base64 string returned by the operator back into binary data for display or for writing to a table.

Step 4: Review the results

Create another cell and run the following code to inspect the schema and contents of the DataFrame:

results

Create one more cell and run the following code to display the first six frames next to their action descriptions in the notebook:

from emrssutils.utils import show_df_image

show_df_image(results, image_row_name='frame_blob', time_row_name='frame_time', description_row_name='scene_description', limit=6)

Each row in the output corresponds to one video frame: the cropped image on the left, and the English action description generated by the model from PROMPT on the right. The cropped images also remain in OUTPUT_DIR, ready to serve as downstream training samples.

What to do next

• Turn the pipeline you prototyped in the notebook into a batch job or submit it to a Ray cluster to process data at a larger scale. For more information, see Submit jobs to a Ray cluster.