EMR Serverless Ray: Integrating AI Inference into Distributed Data Pipelines

Integrating AI Inference into Data Processing with EMR Serverless Ray

Batch workloads often require chaining data preparation, AI inference, and post‑processing: reading road‑test images to detect construction, pedestrians, or bad weather; enriching product images with categories and attributes; transcribing audio recordings for search and analysis. The typical flow is read → preprocess → AI inference → validate → write . When scaling from a few samples to large file batches, the pipeline must also handle concurrent calls, rate‑limit recovery, and result correlation.

Ray Data AI in EMR Serverless Ray

EMR Serverless Ray provides ray.data.ai, which treats model invocation as a standard step in a Python data pipeline. Through the Dataset .ai accessor, tasks such as text summarization, image/video understanding, vectorization, speech‑to‑text, and candidate reranking can be combined with reading, filtering, and writing. The following sections detail the key capabilities and each AI function with runnable examples.

Advantages of AI Functions

Pipeline Integration

Data reading, cleaning, filtering, AI inference, and result writing are chained in a single Python job, handling text, images, video, and audio uniformly. Pre‑processing (e.g., text cleaning, low‑quality image removal, video frame extraction) can precede model calls for summarization, labeling, or vectorization. AI steps compose with existing Dataset transformations and custom logic, leveraging Ray’s distributed data processing to avoid cross‑system exports, result back‑filling, and intermediate file maintenance.

Concurrency and Fault Tolerance

The platform overlaps network and model latency via asynchronous concurrency, increasing batch throughput. It adaptively adjusts concurrency and request pacing based on service feedback, maximizing quota utilization while smoothing traffic. On HTTP 429 (rate limit) responses, it automatically backs off and retries, then gradually restores send rate. Transient network jitter and temporary service errors are retried per policy, reducing task failures. Most scenarios work with defaults, eliminating custom concurrency and retry logic.

Result Association

AI functions preserve original input columns and write results to a specified column, keeping business IDs, raw text, or media URIs alongside model outputs for traceability. Downstream steps can extract content, vectors, or rankings for further filtering, analysis, and writing. For example, product moderation can retain product ID, image URI, and moderation labels for later retrieval and review. When a model request fails but processing should continue, try_ variants (e.g., try_ai_query) set the failed row’s result to None while preserving successful rows.

Usage Statistics

Job logs automatically aggregate model request counts and service‑reported token usage, enabling comparison across models, prompts, and input data. Structured results expose a token_usage() method for in‑code aggregation; raw and rerank usage appears in job logs. Combined with row counts and execution time, these metrics support cost and efficiency analysis. Statistics reflect actual service responses; missing fields remain blank, and final billing follows the model service invoice.

AI Function Reference

All functions share common parameter patterns: col("column") – references per‑row input. lit(value) – supplies a constant for every row. service_name – identifies a registered model service in the workspace. output_col – names the result column; existing columns are retained, same‑named columns are replaced.

The returned Dataset can be further filtered, projected, and written.

Function Overview

ai_query – generation, QA, image/video understanding. Result struct field: content.

ai_embedding – single or array text vectorization. Result struct field: embedding.

ai_embedding_multimodal – image, video, or text vectorization (model‑dependent). Result struct field: embedding.

ai_transcribe – single audio transcription. Result struct field: content.

ai_rerank – per‑row candidate reranking. Result: JSON array string with index and relevance_score.

ai_query – Text Summarization Example

ds = ray.data.read_parquet("oss://<bucket>/corpus/raw_text.parquet")
answers = ds.ai.ai_query(
    service_name="qwen3.7-plus",
    prompt=col("text"),
    output_col="answer",
    options={
        "system_message": "你是文本摘要专家，用一句话概括核心观点。",
        "temperature": 0.3,
    },
)
summaries = answers.select(
    col("id"),
    col("answer").struct["content"].alias("summary"),
    col("answer").struct["error"].alias("error"),
)
summaries.write_parquet("oss://<bucket>/corpus/summarized/")

Input assumes id and text columns. The answer struct contains content (model text), error (failure info), and token usage. content defaults to string; for JSON output, specify format in the prompt and parse downstream.

ai_embedding – Text Vectorization

embeddings = ds.ai.ai_embedding(
    service_name="text-embedding-v4",
    input=col("text"),
    output_col="emb",
)
vectors = embeddings.select(
    col("id"), col("emb").struct["embedding"].alias("vector")
)

Supported models include text-embedding-v4 and qwen3.7-text-embedding. Default input is a single string per row yielding a 1‑D vector; for multiple texts per row, set input_type="string_list" and pass a string array to get a 2‑D vector. None input yields None output.

ai_embedding_multimodal – Image/Video Vectorization

image_index = ray.data.read_parquet("oss://<bucket>/products/image_index.parquet")
image_embeddings = image_index.ai.ai_embedding_multimodal(
    service_name="tongyi-embedding-vision-plus",
    data=col("image_uri"),
    data_type="uri",
    output_col="emb",
)

Result vector at emb.embedding. Each row accepts one media item or text; text input uses data_type="text" (model‑dependent). Vectors for the same retrieval task must use consistent model and configuration.

ai_transcribe – Audio Transcription

audio_index = ray.data.read_parquet("oss://<bucket>/recordings/audio_index.parquet")
transcripts = audio_index.ai.ai_transcribe(
    service_name="qwen-audio-3.0-asr-flash",
    data=col("audio_uri"),
    data_type="uri",
    output_col="transcript",
    options={"language": "zh"},
)

Transcribed text at transcript.content. For translation or summarization, pipe the output to ai_query.

ai_rerank – Candidate Reranking

candidates = ray.data.from_items([{
    "query": "适合通勤的背包",
    "documents": ["轻便双肩电脑包", "大容量旅行箱", "折叠雨伞"],
}])
ranked = candidates.ai.ai_rerank(
    query=col("query"),
    documents=col("documents"),
    service_name="qwen3-vl-rerank",
    options={"top_n": 2, "return_documents": True},
    output_col="ranked",
)

Each row supplies a query and candidate list; output is a JSON array string sorted by relevance, containing original index and relevance_score. Supports text, image, and video candidates per model capability. top_n controls returned count; return_documents=True includes candidate text.

Multimodal Understanding with ai_query

Image and video understanding use models like qwen3.7-plus and qwen3.8-max via service_name. Example: product image categorization.

images = ray.data.read_parquet("oss://<bucket>/products/image_index.parquet")
labels = images.ai.ai_query(
    service_name="qwen3.7-plus",
    prompt=lit("判断图片中的商品类别，并说明判断依据。"),
    data=col("image_uri"),
    data_type="uri",
    output_col="label",
)

Production media inference uses uri (single) or uri_list (multiple images/video frames). binary / binary_list are for functional testing only and not recommended for production due to performance. When reading from file directories, set include_uris=True to output a uri column, then pass data=col("uri") to the AI function.

Single video: data_type="uri" with .mp4, .mov URLs.

videos = ray.data.from_items([{"video_uri": "oss://<bucket>/videos/demo.mp4"}])
video_analysis = videos.ai.ai_query(
    service_name="qwen3.7-plus",
    prompt=lit("概括这段视频的主要内容，并按时间顺序列出关键事件。"),
    data=col("video_uri"),
    data_type="uri",
    output_col="video_analysis",
)
video_analysis.select(
    col("video_uri"),
    col("video_analysis").struct["content"].alias("summary"),
).show()

Pre‑extracted frames: pass a chronologically ordered URI list per row with data_type="uri_list" and options={"list_data_type": "video"}.

video_analysis = video_index.ai.ai_query(
    service_name="qwen3.7-plus",
    prompt=col("question"),
    data=col("frame_uris"),
    data_type="uri_list",
    options={"list_data_type": "video"},
    output_col="video_analysis",
)

Multi‑page documents (e.g., contracts): convert PDF pages to images, store ordered URIs per row, and process as an image list.

contracts = contract_pages.ai.ai_query(
    service_name="qwen3.7-plus",
    prompt=lit("提取合同中的甲方、乙方、金额和有效期，按 JSON 返回。"),
    data=col("page_uris"),
    data_type="uri_list",
    options={"list_data_type": "image"},
    output_col="contract",
)

Result at contract.content; parse JSON and validate fields. Page count and image size must respect service limits.

Practical Cases

Road‑Test Scene Recognition

Filter low‑quality frames (resolution, brightness, contrast) then classify scenes.

INPUT_PATH = "oss://<bucket>/autonomous/camera_frames/"
OUTPUT_PATH = "oss://<bucket>/autonomous/scene_labels/"

images = ray.data.read_images(
    INPUT_PATH,
    include_uris=True,
    file_extensions=["jpg", "jpeg", "png"],
)

def filter_quality(batch: pd.DataFrame) -> pd.DataFrame:
    keep = []
    for image in batch["image"]:
        height, width = image.shape[:2]
        gray = image.mean(axis=2) if image.ndim == 3 else image
        brightness = float(gray.mean())
        contrast = float(gray.var())
        keep.append(
            width >= 1280 and height >= 720
            and 25 <= brightness <= 230
            and contrast >= 100
        )
    return batch.loc[keep].reset_index(drop=True)

qualified = images.map_batches(filter_quality, batch_format="pandas").select_columns(["uri"])

labeled = qualified.ai.ai_query(
    service_name="qwen3.7-plus",
    prompt=lit(
        "判断图片属于哪个道路场景，只返回以下一个标签："
        "NORMAL、PEDESTRIAN_CROSSING、CONSTRUCTION、"
        "EMERGENCY_VEHICLE、ACCIDENT、BAD_WEATHER。"
    ),
    data=col("uri"),
    data_type="uri",
    output_col="scene_label",
)

output = labeled.select(
    col("uri"),
    col("scene_label").struct["content"].alias("label"),
    col("scene_label").struct["error"].alias("error"),
)
output.write_parquet(OUTPUT_PATH)

Quality thresholds and label set are adjustable. The error field aids post‑hoc failure inspection. Pipeline: read images → quality filter → AI scene classification → write URI + label index .

Cooking Key‑Frame Filtering

From extracted cooking video frames, keep only those showing valid cooking actions (pots, ingredient handling, hand movements).

FRAME_INDEX_PATH = "oss://<bucket>/cooking_videos/frame_index/"
OUTPUT_PATH = "oss://<bucket>/cooking_videos/kept_frame_index/"

frames = ray.data.read_parquet(FRAME_INDEX_PATH).select_columns(["uri"])

judged = frames.ai.ai_query(
    service_name="qwen3.7-plus",
    prompt=lit(
        "判断图片是否包含有效的烹饪操作画面，"
        "如锅具、食材处理或手部操作。"
        "包含只返回 KEEP，不包含只返回 DROP。"
    ),
    data=col("uri"),
    data_type="uri",
    output_col="answer",
)

rows = judged.select(
    col("uri"),
    col("answer").struct["content"].alias("decision"),
    col("answer").struct["error"].alias("error"),
)
kept = rows.filter(
    lambda row: row["error"] is None
    and isinstance(row["decision"], str)
    and row["decision"].strip().upper() == "KEEP"
)
kept.write_parquet(OUTPUT_PATH)

Only rows with KEEP and no error are retained. Output is a Parquet index of frame URIs and decisions; original images remain in place.

Model Selection Reference

Text, image & video understanding – ai_query – models: qwen3.7-plus, qwen3.8-max Text vectorization – ai_embedding – models: text-embedding-v4, qwen3.7-text-embedding Image vectorization – ai_embedding_multimodal – model: tongyi-embedding-vision-plus Speech transcription – ai_transcribe – model: qwen-audio-3.0-asr-flash Text & multimodal candidate reranking – ai_rerank – model: qwen3-vl-rerank Additional models can be chosen from the AI Center per business needs.

Failure Handling & Usage Inspection

try_* variants (e.g., try_ai_query, try_ai_embedding) – on failure, the result column is None; other rows continue. Model service configuration or input column issues still require fixing.

_raw variants (e.g., ai_query_raw) – return the full service response as a JSON string for custom parsing.

materialize() – forces execution and caches the Dataset, enabling multiple inspections, statistics, and writes without re‑running inference.

token_usage() – aggregates token usage from structured results; raw and rerank usage appear in job logs.

texts = ray.data.from_items([{"text": "请简述什么是向量检索。"}])
cached = texts.ai.ai_query(
    service_name="qwen3.7-plus",
    prompt=col("text"),
    output_col="answer",
).materialize()

print(cached.ai.token_usage(output_col="answer"))
cached.write_parquet("oss://<bucket>/inference/results/")

Subsequent operations reuse cached. Missing usage fields show as blank; final billing follows the model service invoice.

Extending Pipelines with Custom Logic

Custom pre‑/post‑processing (text cleaning, field computation, local models) can be inserted into the same pipeline. Subclass BaseOperator and implement process_row(), then apply via .ai.apply().

from ray.data.ai import BaseOperator

class TextLength(BaseOperator):
    def process_row(self, row):
        return len(row["text"])

with_length = ds.ai.apply(TextLength, output_col="text_length")

For HTTP‑based steps, use .ai.apply_http().

Common Business Scenarios

Product image moderation & attribute extraction – input: image URI – capability: ai_query Contract & multi‑page document extraction – input: ordered page image URI list – capability: ai_query (multi‑image)

User comment classification & summarization – input: text – capability: ai_query RAG knowledge base vectorization – input: text or segmented text array – capability: ai_embedding Short video understanding & tagging – input: video URI or ordered frame list – capability: ai_query (video)

Image similarity search – input: image URI – capability: ai_embedding_multimodal Audio transcription & downstream summarization – input: audio URI, transcribed text – capabilities: ai_transcribe + ai_query Retrieval candidate reranking – input: query + candidate set – capability: ai_rerank Start with a small dataset (text table, image index, audio list), pick a concrete goal (summary, labels, vectors), run the read‑AI‑write loop, inspect results, failures, usage, and latency, then scale. Once images have labels, audio has transcripts, and text has vectors, downstream filtering, analysis, and search become feasible. Wiring these steps into a single Ray data pipeline turns a one‑off model call into a repeatable business process.