Description #

General video format conversion processor

Key features #

• Supports conversion between multiple video formats
• Automatically selects the appropriate encoder
• Supports custom ffmpeg parameters via extra_params

Format support #

• Input: Mainstream video formats such as AVI, MOV, MKV, FLV, WMV, 3GP, MP4
• Output: MP4, AVI, MOV, MKV, FLV, WEBM, and more
• Video codecs: H.264, H.265, VP8, VP9, and more
• Audio codecs: AAC, MP3, Opus, and more

Operator parameters #

Input #

Output #

Returns an array of output paths on success, or None on failure.

Parameters #

If a parameter does not have a default value, it is required.

Examples #

The following code demonstrates how to use Daft (for distributed scenarios) to run the operator for general video format conversion. Supports conversion to multiple formats, including MP4, AVI, MOV, MKV, WEBM, and more.

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 VideoConvert

if __name__ == "__main__":
    # The converted video will be saved to the specified TOS path. Therefore, you need to set environment variables to ensure write permissions to 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 = {
        "input_path": [f"https://{TOS_TEST_DIR_URL}/public/archive/video_convert/music_sample.mov"],
        "output_path": [f"tos://{TOS_TEST_DIR}/video_convert/music_sample.mp4"],
    }
    ds = daft.from_pydict(samples)

    # Using Daft to convert video format
    constructor_kwargs = {
        "output_format": "mp4",
        "extra_params": ["-crf", "23", "-preset", "medium"],
    }

    ds = ds.with_column(
        "convert_result",
        las_udf(
            VideoConvert,
            construct_args=constructor_kwargs,
            num_cpus=1,
            concurrency=1,
            batch_size=1,
        )(col("input_path"), col("output_path")),
    )

    ds.show()
    # ╭────────────────────────────────┬────────────────────────────────┬────────────────────────────────╮
    # │ input_path                     ┆ output_path                    ┆ convert_result                 │
    # │ ---                            ┆ ---                            ┆ ---                            │
    # │ String                         ┆ String                         ┆ String                         │
    # ╞════════════════════════════════╪════════════════════════════════╪════════════════════════════════╡
    # │ https://las-public-data-qa.to… ┆ tos://tos_bucket/video_conver… ┆ tos://tos_bucket/video_conver… │
    # ╰────────────────────────────────┴────────────────────────────────┴────────────────────────────────╯