advancedView Python source
Find scenes in videos with text search
This example searches inside video by description. The video is a 29-second montage of six real clips, cut together: a sunset at sea, a tram on a snowy street, a waterfall, a black cat, a rocket launch at night, and a hurricane seen from space. Inside the engine the example keeps one frame per second and embeds each frame with CLIP. SQL then splits the video into scenes and ranks them for queries such as "a black cat", returning each match with its start and end time.
Run it
Run all commands below from the repository root (sail-examples/):
uv run multimodal/search-video-frames/main.py
The first run downloads the two CLIP ViT-B/32 encoders in ONNX format (about 600 MB together) from Hugging Face into the local Hugging Face cache. No account or key is needed. The model revisions are pinned.
The run finds all six scenes, and each query's best match is the scene that shows what the query describes. Times are whole seconds, because the example samples one frame per second: the rocket launch, which runs from 19.2 to 24.2 seconds in the video, comes back as 20 s to 25 s.
Output
The newest capture appears first. Older output stays here when a result changes.
Recorded output · Sail 0.7.3
Captured on Sail 0.7.3. Latest capture: 2026-11-05, Python 3.12.14.
The example has changed since this capture. Rerun it to refresh the output.
+------------------+-----+-----+---+------+-----------------+------------------------+
| video|scene|start|end|frames|similarity_at_cut|lowest_similarity_within|
+------------------+-----+-----+---+------+-----------------+------------------------+
|scenes-montage.mp4| 1| 0| 5| 5| NULL| 0.949|
|scenes-montage.mp4| 2| 5| 10| 5| 0.498| 0.922|
|scenes-montage.mp4| 3| 10| 15| 5| 0.465| 0.977|
|scenes-montage.mp4| 4| 15| 20| 5| 0.538| 0.925|
|scenes-montage.mp4| 5| 20| 25| 5| 0.408| 0.968|
|scenes-montage.mp4| 6| 25| 30| 5| 0.635| 0.98|
+------------------+-----+-----+---+------+-----------------+------------------------+
'a rocket launch at night' #1: scenes-montage.mp4 20s to 25s score 0.290 (the scene shows: a rocket lifting off from its launch pad at night)
'a rocket launch at night' #2: scenes-montage.mp4 0s to 5s score 0.215 (the scene shows: an orange sun low over a calm sea with a small boat)
'a black cat' #1: scenes-montage.mp4 15s to 20s score 0.300 (the scene shows: a black cat lying on a light-colored carpet)
'a black cat' #2: scenes-montage.mp4 0s to 5s score 0.210 (the scene shows: an orange sun low over a calm sea with a small boat)
'a waterfall' #1: scenes-montage.mp4 10s to 15s score 0.291 (the scene shows: a waterfall pouring over a rock ledge into a pool)
'a waterfall' #2: scenes-montage.mp4 0s to 5s score 0.204 (the scene shows: an orange sun low over a calm sea with a small boat)
'a tram on a snowy street' #1: scenes-montage.mp4 5s to 10s score 0.332 (the scene shows: a tram approaching along tracks on a snowy city street)
'a tram on a snowy street' #2: scenes-montage.mp4 15s to 20s score 0.129 (the scene shows: a black cat lying on a light-colored carpet)
Python code
main.py260 lines
# /// script
# requires-python = ">=3.12"
# dependencies = [
# "certifi==2026.10.19",
# "pysail==0.7.3",
# "pyspark-client==4.1.3",
# "pyarrow==26.0.0",
# "pandas==3.0.6",
# "numpy==2.5.4",
# "av==18.0.0",
# "onnxruntime==1.31.1",
# "tokenizers==0.22.1",
# "huggingface-hub==0.36.0",
# "pillow==13.0.0",
# ]
# ///
"""Find the scenes in videos that match a text query, with CLIP inside the engine.
The pattern: a table of video paths goes through `mapInArrow`, which decodes each video, keeps
one frame per second, and returns one CLIP vector per kept frame (one video in, many small rows
out). SQL window functions then split each video into scenes where consecutive frames stop
looking alike, and rank the scenes for each text query.
"""
import atexit
import shutil
import tempfile
from pathlib import Path
import pyarrow as pa
from pyspark.sql import DataFrame, SparkSession, Window
from pyspark.sql import functions as F
# CLIP ViT-B/32, with the image and text encoders exported to ONNX as two models.
VISION_REPO = "Qdrant/clip-ViT-B-32-vision"
VISION_REVISION = "e0c24ed0fa57fa3e4f97f30de74c51d944036ace" # pinned: every run uses the same weights
TEXT_REPO = "Qdrant/clip-ViT-B-32-text"
TEXT_REVISION = "48ca1db27cb4063eb311ec2aa7f087a808112876"
CONTEXT_LENGTH = 77 # CLIP reads at most 77 tokens
INFERENCE_BATCH_SIZE = 16 # bound model tensors independently of Arrow input batch size
SCENE_CHANGE_BELOW = 0.8 # a new scene starts when a frame is less similar than this to the one before
QUERIES = ["a rocket launch at night", "a black cat", "a waterfall", "a tram on a snowy street"]
VIDEOS = Path(__file__).resolve().parent / "videos" # sources and licenses: videos/CREDITS.md
FRAMES_PER_CHUNK = 32 # frames per model call and per output batch
FRAME_COLUMNS = "video string, second int, embedding array<float>"
FRAME_SCHEMA = pa.schema(
[("video", pa.string()), ("second", pa.int32()), ("embedding", pa.list_(pa.float32()))]
)
TEXT_COLUMNS = "query_id long, embedding array<float>"
TEXT_SCHEMA = pa.schema([("query_id", pa.int64()), ("embedding", pa.list_(pa.float32()))])
def temp_dir(prefix: str) -> str:
"""A temporary directory that is removed when the process exits."""
path = tempfile.mkdtemp(prefix=prefix)
atexit.register(shutil.rmtree, path, ignore_errors=True)
return path
def unit_rows(vectors):
"""Scale each row to length 1, so cosine similarity is a dot product."""
import numpy as np
return vectors / np.clip(np.linalg.norm(vectors, axis=1, keepdims=True), 1e-12, None)
def embed_frames(batches):
"""mapInArrow function: one video in, one row per kept frame out, in chunks."""
import av
import numpy as np
import onnxruntime as ort
from huggingface_hub import hf_hub_download
from PIL import Image
# Download where the task runs (cached per machine), so this also works on cluster workers.
model_path = hf_hub_download(VISION_REPO, "model.onnx", revision=VISION_REVISION)
options = ort.SessionOptions()
options.intra_op_num_threads = 1 # the engine runs partitions in parallel already
session = ort.InferenceSession(model_path, sess_options=options, providers=["CPUExecutionProvider"])
mean = np.array([0.48145466, 0.4578275, 0.40821073], dtype=np.float32)
std = np.array([0.26862954, 0.26130258, 0.27577711], dtype=np.float32)
def preprocess(image):
# CLIP's preprocessing: shortest side to 224 (bicubic), center crop 224, normalize.
# The shorter side becomes exactly 224; the longer side is scaled and truncated, as transformers does.
short, long = sorted(image.size)
resized = (224, int(224 * long / short))
image = image.resize(resized if image.width <= image.height else resized[::-1], Image.BICUBIC)
left, top = (image.width - 224) // 2, (image.height - 224) // 2
image = image.crop((left, top, left + 224, top + 224))
return ((np.asarray(image, dtype=np.float32) / 255.0 - mean) / std).transpose(2, 0, 1)
def embedded(video, seconds, pixels):
(vectors,) = session.run(["image_embeds"], {"pixel_values": np.stack(pixels)})
rows = {"video": [video] * len(seconds), "second": seconds, "embedding": unit_rows(vectors).tolist()}
return pa.RecordBatch.from_pydict(rows, schema=FRAME_SCHEMA)
for batch in batches:
for path in batch.column("path").to_pylist():
video, seconds, pixels, next_second = Path(path).name, [], [], 0
with av.open(path) as container:
stream = container.streams.video[0]
origin = stream.start_time or 0
for frame in container.decode(stream):
if frame.pts is None:
continue
# Exact time since the stream started, as a fraction of seconds (no float rounding).
elapsed = (frame.pts - origin) * stream.time_base
if elapsed < next_second:
continue
# Keep the first frame at or after each whole second, labeled with its second.
second = int(elapsed)
seconds.append(second)
pixels.append(preprocess(frame.to_image()))
next_second = second + 1
# Embed and emit fixed-size chunks, so memory and batches stay small however long a
# video is.
if len(pixels) == FRAMES_PER_CHUNK:
yield embedded(video, seconds, pixels)
seconds, pixels = [], []
if pixels:
yield embedded(video, seconds, pixels)
def embed_texts(batches):
"""mapInArrow function for the text encoder."""
import numpy as np
import onnxruntime as ort
from huggingface_hub import hf_hub_download
from tokenizers import Tokenizer
model_path = hf_hub_download(TEXT_REPO, "model.onnx", revision=TEXT_REVISION)
tokenizer_path = hf_hub_download(TEXT_REPO, "tokenizer.json", revision=TEXT_REVISION)
options = ort.SessionOptions()
options.intra_op_num_threads = 1
session = ort.InferenceSession(model_path, sess_options=options, providers=["CPUExecutionProvider"])
tokenizer = Tokenizer.from_file(tokenizer_path)
tokenizer.enable_truncation(CONTEXT_LENGTH)
# Fixed-length padding makes a row's vector independent of the other rows in its batch.
end_of_text = "<|endoftext|>"
tokenizer.enable_padding(
length=CONTEXT_LENGTH, pad_id=tokenizer.token_to_id(end_of_text), pad_token=end_of_text
)
for input_batch in batches:
for start in range(0, input_batch.num_rows, INFERENCE_BATCH_SIZE):
batch = input_batch.slice(start, INFERENCE_BATCH_SIZE)
texts = batch.column("text").to_pylist()
if not texts:
continue
encodings = tokenizer.encode_batch(texts)
feeds = {
"input_ids": np.array([e.ids for e in encodings], dtype=np.int64),
"attention_mask": np.array([e.attention_mask for e in encodings], dtype=np.int64),
}
(vectors,) = session.run(["text_embeds"], feeds)
yield pa.RecordBatch.from_pydict(
{"query_id": batch.column("query_id"), "embedding": unit_rows(vectors).tolist()},
schema=TEXT_SCHEMA,
)
def run(spark: SparkSession) -> dict[str, DataFrame]:
# What each scene really shows, to check the results by eye.
truth = spark.read.csv(
str(VIDEOS / "scenes.csv"), header=True, schema="video string, start double, end double, shows string"
)
# A table of paths, one row per video. Sample and embed the frames, then save the vectors once.
frames_path = temp_dir("frames-")
paths = sorted(str(p) for p in VIDEOS.glob("*.mp4"))
videos = spark.createDataFrame([(p,) for p in paths], "path string").repartition(2)
videos.mapInArrow(embed_frames, FRAME_COLUMNS).write.mode("overwrite").parquet(frames_path)
frames = spark.read.parquet(frames_path)
# Scenes: a frame that looks unlike the one before it starts a new scene.
in_order = Window.partitionBy("video").orderBy("second")
previous = F.lag("embedding").over(in_order)
similarity = F.call_function("vector_cosine_similarity", F.col("embedding"), previous)
new_scene = F.when(previous.isNull() | (similarity < SCENE_CHANGE_BELOW), 1).otherwise(0)
numbered = (
frames.withColumn("similarity", similarity.cast("double"))
.withColumn("new_scene", new_scene)
.withColumn(
"scene",
F.sum("new_scene").over(in_order.rowsBetween(Window.unboundedPreceding, Window.currentRow)),
)
)
scenes = numbered.groupBy("video", "scene").agg(
F.min("second").alias("start"),
(F.max("second") + 1).alias("end"),
F.count("*").alias("frames"),
# How alike consecutive sampled frames are: across the cut that starts the scene (null for the
# first scene), and the lowest within the scene. Use these to choose SCENE_CHANGE_BELOW.
F.max(F.when(F.col("new_scene") == 1, F.col("similarity"))).alias("similarity_at_cut"),
F.min(F.when(F.col("new_scene") == 0, F.col("similarity"))).alias("lowest_similarity_within"),
)
# Search: a scene's score for a query is its best frame's score.
queries = spark.createDataFrame(list(enumerate(QUERIES)), "query_id long, text string")
query_vectors = queries.mapInArrow(embed_texts, TEXT_COLUMNS).withColumnRenamed("embedding", "text_vec")
scored = query_vectors.crossJoin(numbered.withColumnRenamed("embedding", "frame_vec")).select(
"query_id",
"video",
"scene",
F.call_function("vector_cosine_similarity", F.col("text_vec"), F.col("frame_vec"))
.cast("double")
.alias("score"),
)
best = scored.groupBy("query_id", "video", "scene").agg(F.max("score").alias("score"))
rank = F.row_number().over(Window.partitionBy("query_id").orderBy(F.desc("score"), "video", "scene"))
# A found scene shows what the true scene that contains its first sampled second shows.
shows = (
scenes.alias("s")
.join(
truth.alias("t"),
(F.col("s.video") == F.col("t.video"))
& (F.col("s.start") >= F.col("t.start"))
& (F.col("s.start") < F.col("t.end")),
"left",
)
.select("s.video", "s.scene", "t.shows")
)
top2 = (
best.withColumn("rank", rank)
.filter("rank <= 2")
.join(queries, "query_id")
.join(scenes, ["video", "scene"])
.join(shows, ["video", "scene"], "left")
.select("query_id", "text", "rank", "video", "start", "end", "score", "shows")
)
return {"scenes": scenes.orderBy("video", "scene"), "search": top2.orderBy("query_id", "rank")}
if __name__ == "__main__":
from pysail.spark import SparkConnectServer
server = SparkConnectServer()
server.start()
_, port = server.listening_address
spark = SparkSession.builder.remote(f"sc://localhost:{port}").getOrCreate()
results = run(spark)
results["scenes"].select(
"video",
"scene",
"start",
"end",
"frames",
F.round("similarity_at_cut", 3).alias("similarity_at_cut"),
F.round("lowest_similarity_within", 3).alias("lowest_similarity_within"),
).show()
for row in results["search"].collect():
print(
f"{row['text']!r} #{row['rank']}: {row['video']} {row['start']}s to {row['end']}s"
f" score {row['score']:.3f} (the scene shows: {row['shows']})"
)
spark.stop()
server.stop()
How it works
- A table of paths. The videos in
multimodal/search-video-frames/videos/are listed in a DataFrame with onepathper file. Each task opens only the files in its own rows. - One video in, many rows out.
mapInArrowdownloads the image encoder where the task runs (cached per machine) and loads it once per partition. For each video it decodes the frames with PyAV and keeps the first frame at or after each whole second, measured from the start of the video stream. It returns one row per kept frame: the video, the second and anARRAY<FLOAT>vector. Frames are embedded and returned at most 32 at a time. This bounds frames per model call and output batch; memory also depends on frame resolution, model size and concurrent partitions. - Save once. The frame vectors are written to Parquet and read back, so later queries never decode or embed the videos again.
- Split into scenes in SQL.
lagover each video's sampled frames, in time order, gives the previous sampled frame's vector. Whenvector_cosine_similaritybetween consecutive sampled frames (one per second) drops below 0.8 (SCENE_CHANGE_BELOW), a new scene starts. A runningsumof those starts numbers the scenes, and agroupBygives each scene's start and end second. Each scene also reportssimilarity_at_cut, the similarity across the cut that starts it (null for a video's first scene), andlowest_similarity_within, the lowest similarity between consecutive sampled frames inside it. - Search in SQL. Each query is embedded with CLIP's text encoder, every frame is scored with
vector_cosine_similarity, and a scene's score is its best frame's score. A window function keeps the two best scenes per query.
multimodal/search-video-frames/videos/scenes.csv records where each clip really starts and ends. The output joins it in, so you can check the matches by eye.
Use your own videos
Put your files in multimodal/search-video-frames/videos/, or build the videos DataFrame from any list of files, and change QUERIES. PyAV's wheels are linked against FFmpeg (per the PyAV README), so it opens the formats that FFmpeg build supports. The paths must be readable from wherever the UDF runs: local paths on one machine, object storage on a cluster (open the file with a client such as fsspec and pass the file object to av.open). A file that cannot be decoded raises an error and fails the job, and a video with no decodable frames returns no rows. To record such files and keep going, catch the error per file and return a status column, as the OCR example does.
On this video, the scenes output shows consecutive sampled frames (one per second) at least 0.92 similar within each scene, and at most 0.65 similar across each cut, so 0.8 separates them. Footage with fast camera motion or gradual transitions can fall between those values, so check the two similarity columns on a few of your own videos and adjust SCENE_CHANGE_BELOW. For long videos, keeping one frame every few seconds reduces the work per video.
Video
Four clips are from Wikimedia Commons under CC0, and two are NASA videos. videos/CREDITS.md lists each clip's source, creator and license.