Dual‑Dimension Cost Cutting for EMR Serverless Spark AI Functions

Cost Model

Task total cost ≈ model inference cost + Spark compute cost. Model inference cost ≈ effective calls × average tokens per call × model unit price. Spark compute cost ≈ active resource size × active duration × resource unit price.

Dimension 1 – AI Query Optimization

Filter push‑down

When a filter does not depend on AI results, the optimizer moves the filter before the AI function so that only rows surviving the filter are sent to the model. Example:

SELECT normalized_text, sentiment
FROM (
  SELECT upper(review_text) AS normalized_text,
         ai_sentiment(review_text) AS sentiment
  FROM reviews
) enriched
WHERE normalized_text LIKE '%REFUND%';

If the filter leaves 20 % of rows, AI calls drop to roughly 20 %.

Limit and ordering

For queries without ordering, Spark stops after enough rows, but the optimizer can determine the exact rows needed before invoking the model. For ordered limits where the order does not depend on AI output, the engine sorts and truncates first, then calls the model for the final N rows. In a LIMIT 10 query the optimizer moves GlobalLimit after AIProject, reducing AI requests from 26 to 10, token usage from 65,572 to 25,220, and request latency from 1.6 min to 28.8 s.

Input deduplication

When many rows contain identical inputs, the engine computes the AI result once and reuses it for all duplicates. This is effective when the duplicate rate is high and the per‑call cost is significant.

Overall optimizer

The optimizer applies these and additional rules jointly, always preserving SQL semantics while minimizing data sent to AI nodes.

Dimension 2 – Asynchronous Batch File Inference

Batch File is an Alibaba Cloud Bailei API for non‑real‑time large‑scale inference. It submits a batch request file, processes it asynchronously, and returns results later. The unit price is roughly 50 % of the real‑time price, and executors can be released during the waiting period.

Execution phases

Submit : Spark reads input, creates a batch request file and sends it to the model service.

Wait : After submission the executor is released; a lightweight coordinator tracks batch status.

Recover : When the model finishes, Spark pulls the results, aligns them with the original rows, and continues downstream processing.

This decouples remote waiting time from Spark compute time, allowing resource shrinkage for offline jobs.

Combined use‑case – 5 M review labeling

A raw pipeline would invoke three AI functions on every row (≈ 15 M calls). Query optimization (date and category filters) reduces the candidate set to 1.2 M rows; input deduplication reduces unique inputs to 850 k. The final AI call count becomes 255 k (≈ 17 % of the baseline). Enabling Batch File halves the model unit price, yielding a relative inference cost of 8.5 % of the original real‑time cost (≈ 91.5 % reduction).

Performance numbers for a LIMIT 10 query: AI requests 26 → 10, tokens 65,572 → 25,220, latency 1.6 min → 28.8 s.

Configuration and tuning

SET spark.emr.serverless.ai.batchFile.enabled = true;
SET spark.emr.serverless.ai.batchFile.mode = async;
SET spark.emr.serverless.ai.deduplicate.enabled = true;
SET spark.emr.serverless.ai.batchFile.maxRequestsPerFile = 10000;
SET spark.emr.serverless.ai.batchFile.completionWindow = 24h;

Per‑function mode can be set via the options argument, e.g.:

SELECT review_id,
       ai_sentiment(review_text,
                    options => '{"batch_mode":"async"}') AS sentiment
FROM reviews;

Trade‑offs and selection guidance

Online AI Function + query optimization – suitable for low‑latency, interactive or streaming workloads; query optimization reduces calls for any execution mode.

Asynchronous Batch File + query optimization – suitable for large, cost‑sensitive offline jobs with clear batch boundaries; batch file lowers model price and allows executor shrinkage.

Batch File does not support multi‑modal inputs (e.g., images); such calls fall back to online mode.

Practical recommendations

Apply AI query optimization to all workloads to eliminate unnecessary model calls. Enable Batch File only when the job can tolerate delayed results and when releasing executors yields cost savings.