Realtime Compute for Apache Flink:Pemanggilan umum

Parameter tipe konten

• Untuk masukan satu kolom, gunakan content-type untuk menentukan tipe konten. Nilai yang didukung: text, image_url.

• Untuk masukan multi-kolom, gunakan content-types untuk menentukan tipe konten setiap kolom. Nilai yang didukung dan pemetaannya ke tipe kolom Flink SQL tercantum di bawah ini:

• Kami menyarankan Anda menentukan content-types dalam parameter. Jika Anda menetapkan keduanya content-type dan content-types, framework akan melaporkan error. Catatan: pengaturan content-type dan content-types dalam parameter model dan pengaturan per-pemanggilan saling memengaruhi:

Parameter tingkat kolom

Pemrosesan bundel

• Pemrosesan: Pemrosesan bundel pertama-tama menggabungkan beberapa catatan masukan menjadi satu batch, lalu memanggil model AI untuk batch tersebut. Hal ini mengurangi jumlah pemanggilan model tetapi meningkatkan waktu inferensi model untuk setiap permintaan. Ketika catatan dapat digabungkan secara efektif, pemrosesan bundel dapat meningkatkan throughput operator secara signifikan.

• Deskripsi parameter: bundle-size menentukan jumlah maksimum catatan yang dikumpulkan dalam satu batch. bundle-allow-latency menentukan waktu tunggu maksimum ketika batch belum mencapai nilai bundle-size. Pemrosesan dimulai ketika salah satu kondisi ukuran batch atau waktu tunggu terpenuhi.

• Persyaratan: Pemrosesan bundel berlaku berdasarkan skema masukan dan keluaran model.

  • Generasi teks: Untuk model yang memiliki task chat/completions, masukan berisi tepat satu kolom STRING berupa teks, dan keluaran berisi tepat satu kolom STRING atau VARIANT. Masukan multi-kolom dan masukan multimodal seperti gambar, video, dan audio tidak didukung.

  • Text embedding: Untuk model yang memiliki task embeddings, masukan berisi tepat satu kolom STRING, dan keluaran berisi tepat satu kolom ARRAY<FLOAT>.

• Fungsi yang didukung: ML_PREDICT, klasifikasi teks (AI_CLASSIFY), analisis sentimen (AI_SENTIMENT), ekstraksi informasi (AI_EXTRACT), Ringkasan teks (AI_SUMMARIZE), translation (AI_TRANSLATE), penyamaran data (AI_MASK), dan text embedding (AI_EMBED). Di antaranya, ML_PREDICT mendukung pemrosesan bundel hanya jika model yang terdaftar memenuhi persyaratan di atas.

• Konfigurasi global: Konfigurasi ini berlaku untuk pemanggilan ML_PREDICT dan pemanggilan fungsi AI spesifik domain yang tercantum dalam pekerjaan yang memenuhi persyaratan di atas. Gunakan parameter global berikut untuk mengaktifkan pemrosesan bundel:

  -- Pemrosesan bundel hanya didukung di VVR 11.9 dan versi lebih baru.
  SET 'table.exec.ml-predict.bundle-size' = '10';
  SET 'table.exec.ml-predict.bundle-allow-latency' = '5s';

• Konfigurasi per-pemanggilan: Parameter ini hanya berlaku untuk pemanggilan fungsi tempat parameter tersebut ditentukan. ML_PREDICT dan fungsi AI spesifik domain yang tercantum menggunakan metode konfigurasi yang sama. Untuk parameter dan contoh, lihat Konfigurasi per-pemanggilan.

Batasan

• Hanya didukung di VVR 11.9 dan versi lebih baru.

• Anda tidak dapat mengaktifkan penanganan error saat pemanggilan model berjalan dalam mode sinkron.

• Penanganan error tidak dapat digunakan bersamaan dengan Vectorization result cache (CACHE_TABLE) saat ini.

Aktifkan penanganan error

Tambahkan parameter FAIL_ON_ERROR => FALSE saat Anda memanggil fungsi AI.

Prasyarat: Pertahankan pengaturan retry default model: error-handling-strategy='retry' dan retry-fallback-strategy='failover'. Untuk informasi selengkapnya tentang parameter model, lihat Pengaturan model.

Perubahan output

Setelah Anda mengaktifkan penanganan error, sistem menambahkan kolom error err_msg bertipe VARIANT nullable di akhir bidang output asli. Kolom ini dihasilkan oleh sistem. Anda tidak perlu mendeklarasikannya di tabel masukan atau dalam klausa OUTPUT CREATE MODEL.

• Pemanggilan berhasil: err_msg bernilai NULL.

• Pemanggilan akhirnya gagal: semua bidang hasil model bernilai NULL, dan err_msg berisi informasi error.

Jika Anda menggunakan SELECT *, hasilnya mencakup kolom error baru. Jika Anda memilih bidang secara eksplisit, Anda harus menentukan kolom tersebut secara manual. Nama kolom default adalah err_msg. Jika nama tersebut bentrok dengan kolom output yang sudah ada, sistem akan menambahkan sufiks numerik, seperti err_msg0. Gunakan nama kolom aktual untuk mereferensinya.

Contoh

CREATE TEMPORARY VIEW input_data AS
SELECT *
FROM (VALUES
  (1, CAST('xxxx (a description of violent conflict that violates humanity)' AS STRING), 'negative'),
  (2, CAST('The customer service of this store has always been among the best' AS STRING), 'positive')
) AS t(id, input_text, expected_label);

CREATE TEMPORARY VIEW ai_results AS
SELECT i.id, i.input_text, i.expected_label,
       r.score, r.label, r.confidence, r.err_msg
FROM input_data AS i,
LATERAL TABLE(AI_SENTIMENT(
  MODEL => MODEL ai_sentiment_model,
  INPUT => i.input_text,
  FAIL_ON_ERROR => FALSE,
  CONFIG => MAP['async', 'true']
)) AS r(score, label, confidence, err_msg);

SELECT * FROM ai_results;

message: 
java.util.concurrent.TimeoutException: AsyncPredictFunction did not complete within the configured timeout. Increase the ML_PREDICT timeout or check external service connectivity.

stack_trace:
java.util.concurrent.TimeoutException: AsyncPredictFunction did not complete within the configured timeout. Increase the ML_PREDICT timeout or check external service connectivity.
at org.apache.flink.table.runtime.operators.predict.AsyncMLPredictRunner.timeout(AsyncMLPredictRunner.java:153)
at org.apache.flink.table.runtime.operators.predict.AsyncMLPredictRunner.timeout(AsyncMLPredictRunner.java:51)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator$KeyedResultHandler.timerTriggered(TableKeyedAsyncWaitOperator.java:462)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator$KeyedResultHandler.lambda$registerTimeout$1(TableKeyedAsyncWaitOperator.java:457)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator.lambda$registerTimer$4(TableKeyedAsyncWaitOperator.java:486)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:2285)
at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$deferCallbackToMailbox$28(StreamTask.java:2276)
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:115)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:415)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:384)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:369)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:1273)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:1217)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:1133)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:1110)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:914)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:716)
at java.base/java.lang.Thread.run(Thread.java:991)

Cara penanganan setiap jenis error

Perilaku berikut berlaku saat penanganan error diaktifkan dan pengaturan retry default dipertahankan. Baris error adalah baris yang menyimpan masukan, mengembalikan hasil model kosong, dan mencatat alasan kegagalan dalam err_msg.

Teks

Contoh berikut mendaftarkan dan menggunakan model bawaan Flink AI Service untuk mengklasifikasikan teks masukan berdasarkan sentimen. Contoh 1 menggunakan parameter tingkat MODEL. Contoh 2 mengganti user-prompt saat pemanggilan.

-- Mendaftarkan model bawaan. Saat content-type tidak ditentukan, nilai default-nya adalah text
CREATE MODEL sentiment_model
INPUT (prompt STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-flash',
  'system-prompt' = 'You are a sentiment classifier. Output one label: negative, positive, or neutral'
);

-- Membuat tabel sumber: data ulasan produk simulasi
CREATE TEMPORARY VIEW input_table(id, content)
AS VALUES
  (1, 'Great quality, soft fabric, fits perfectly'),
  (2, 'Had loose threads on arrival, faded badly after one wash'),
  (3, 'Received the item, looks as pictured'),
  (4, 'Started pilling after two weeks, customer service refused returns'),
  (5, 'Flattering fit, color is even better than the photo, already ordered a third one');

-- Membuat tabel hasil
CREATE TEMPORARY TABLE output_table (
  id INT,
  content STRING,
  sentiment STRING
) WITH (
  'connector' = 'print'
);

-- Menggunakan ML_PREDICT untuk inferensi real-time
-- Contoh 1: Menggunakan parameter tingkat MODEL
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

-- Contoh 2: Menentukan parameter per-pemanggilan
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content),
  MAP['user-prompt', 'Reply in Chinese']);

Output Contoh 1:

Output Contoh 2:

Gambar

Contoh berikut mendaftarkan model multimodal dan mengklasifikasikan gambar masukan.

Kolom masukan gambar (saat content-type adalah image_url) bertipe STRING dan mendukung format berikut:

Jika gambar disimpan dalam kolom Blob dari Paimon Catalog, Anda dapat menggunakan fitur Paimon Blob presigned URL untuk menghasilkan URL publik sebagai masukan. Untuk informasi selengkapnya, lihat Apache Paimon.

-- Mendaftarkan model bawaan dengan content-type diatur ke image_url
CREATE TEMPORARY MODEL sentiment_model
INPUT (prompt STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-flash',
  'content-type' = 'image_url'
);

-- Menggunakan ML_PREDICT untuk inferensi real-time
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

Video

Contoh berikut mendaftarkan model multimodal untuk memahami video masukan.

CREATE TEMPORARY MODEL sentiment_model
INPUT (video_url STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-plus',
  'content-types' = 'video_url',
  'system-prompt' = 'Briefly describe the video content'
);

CREATE TEMPORARY VIEW input_table(content)
AS VALUES ('https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20251111/ckgcxt/%E5%8D%95%E4%BA%BA%E8%AF%B4%E8%AF%9D-1.mp4');

SELECT
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

Teks + gambar tunggal

Contoh berikut mendaftarkan model multimodal untuk inferensi berbasis chat pada masukan teks dan gambar.

CREATE MODEL vl_model
INPUT (text_input STRING, image_input STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;image_url'
);

INSERT INTO result_sink
SELECT content FROM TABLE(ML_PREDICT(
  TABLE image_source,
  MODEL vl_model,
  DESCRIPTOR(text_input, image_input)
));         

Teks + beberapa gambar (beberapa kolom)

Teruskan setiap gambar melalui kolom INPUT terpisah. Tentukan image_url untuk setiap kolom gambar dalam content-types.

CREATE MODEL vl_model_multi
INPUT (prompt STRING, img1 STRING, img2 STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;image_url;image_url'
);

SELECT content FROM TABLE(ML_PREDICT(
  TABLE my_source,
  MODEL vl_model_multi,
  DESCRIPTOR(prompt, img1, img2)
));          

Teks + beberapa gambar (array)

Teruskan beberapa gambar dalam kolom ARRAY<STRING>. Gunakan multi_image_urls dalam content-types.

CREATE MODEL vl_model_array
INPUT (prompt STRING, images ARRAY<STRING>)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;multi_image_urls'
);

SELECT content FROM TABLE(ML_PREDICT(
  TABLE my_source,
  MODEL vl_model_array,
  DESCRIPTOR(prompt, images)
)); 

Teks + video

CREATE TEMPORARY MODEL sentiment_model
INPUT (video_url STRING, question STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-plus',
  'content-types' = 'video_url;text'
);

CREATE TEMPORARY VIEW input_table(content, question)
AS VALUES ('https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20251111/ckgcxt/%E5%8D%95%E4%BA%BA%E8%AF%B4%E8%AF%9D-1.mp4', 'How many people are there?');

SELECT
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content, question));