Operator ID: daft.las.functions.video.video_concat.VideoConcat

Description #

Video concatenation processor that sequentially merges multiple videos into a single output video

Key features: #

• Supports concatenation of multiple video files
• Supports local, TOS, and HTTP path inputs
• Supports configuration of video and audio encoding parameters
• Suitable for batch video merging operations

Format support: #

• MP4 (.mp4)
• MOV (.mov)
• MKV (.mkv)
• AVI (.avi)
• Other common video formats

Operator parameters #

Input #

Output #

Column containing concatenated video paths

Parameters #

Examples #

The following code demonstrates how to use Daft (for distributed scenarios) to run the operator for video concatenation. Both input and output use TOS paths, so you need to set environment variables to ensure permission to read and write TOS, including: ACCESS_KEY, SECRET_KEY, TOS_ENDPOINT, TOS_REGION, TOS_TEST_DIR

from __future__ import annotations

import os

import daft
from daft import col
from daft.las.functions.udf import las_udf
from daft.las.functions.video import VideoConcat

if __name__ == "__main__":
    # Both input and output use TOS paths, so you need to set environment variables to ensure permission to read and write TOS, including: ACCESS_KEY, SECRET_KEY, TOS_ENDPOINT, TOS_REGION, TOS_TEST_DIR
    TOS_TEST_DIR_URL = os.getenv("TOS_TEST_DIR_URL", "las-cn-beijing-public-online.tos-cn-beijing.volces.com")
    TOS_TEST_DIR = os.getenv("TOS_TEST_DIR", "tos_bucket")

    if os.getenv("DAFT_RUNNER", "native") == "ray":
        import logging

        import ray

        def configure_logging():
            logging.basicConfig(
                level=logging.INFO,
                format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
                datefmt="%Y-%m-%d %H:%M:%S",
            )
            logging.getLogger("tracing.span").setLevel(logging.WARNING)
            logging.getLogger("daft_io.stats").setLevel(logging.WARNING)
            logging.getLogger("DaftStatisticsManager").setLevel(logging.WARNING)
            logging.getLogger("DaftFlotillaScheduler").setLevel(logging.WARNING)
            logging.getLogger("DaftFlotillaDispatcher").setLevel(logging.WARNING)

        ray.init(dashboard_host="0.0.0.0", runtime_env={"worker_process_setup_hook": configure_logging})
        daft.set_runner_ray()

    daft.set_execution_config(actor_udf_ready_timeout=600)
    daft.set_execution_config(min_cpu_per_task=0)

    samples = {
        "video_paths": [
            [
                f"https://{TOS_TEST_DIR_URL}/public/shared_video_dataset/music_sample.mp4",
                f"https://{TOS_TEST_DIR_URL}/public/shared_video_dataset/music_sample.mp4",
            ]
        ],
        "output_path": [f"tos://{TOS_TEST_DIR}/video/video_concat/concatenated_video.mp4"],
    }
    ds = daft.from_pydict(samples)

    constructor_kwargs = {
        "output_format": "mp4",
        "video_codec": "libx264",
        "audio_codec": "aac",
    }

    ds = ds.with_column(
        "concat_path",
        las_udf(VideoConcat, construct_args=constructor_kwargs)(
            col("video_paths"),
            col("output_path"),
        ),
    )

    ds.show()
    # ╭────────────────────────────────┬────────────────────────────────┬────────────────────────────────╮
    # │ video_paths                    ┆ output_path                    ┆ concat_path                    │
    # │ ---                            ┆ ---                            ┆ ---                            │
    # │ List[String]                   ┆ String                         ┆ String                         │
    # ╞════════════════════════════════╪════════════════════════════════╪════════════════════════════════╡
    # │ [https://las-cn-beijing-pub…   ┆ tos://tos_bucket/video/video_… ┆ tos://tos_bucket/video/video_… │
    # ╰────────────────────────────────┴────────────────────────────────┴────────────────────────────────╯