Realtime Compute for Apache Flink:Genel çağrı

CREATE TEMPORARY TABLE text_source (
  user_input STRING
) WITH ('connector' = 'datagen');

CREATE TEMPORARY TABLE result_sink (
  user_input STRING,
  ai_analysis STRING
) WITH ('connector' = 'print');

CREATE TEMPORARY MODEL text_model
INPUT (user_input STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.6-flash',
  'task' = 'chat/completions',
  'system-prompt' = 'Rate the gibberish level of the input on a scale of 0 to 100'
);

INSERT INTO result_sink
SELECT user_input, content as ai_analysis FROM
    ML_PREDICT(
        TABLE text_source,
        MODEL text_model,
        DESCRIPTOR(user_input)
);

ML_PREDICT(TABLE <table_name>, MODEL <model_name>, DESCRIPTOR(<input_columns>) [, CONFIG => MAP[...]])        

İçerik türü parametreleri

• Tek sütunlu girişler için içerik türünü belirtmek üzere content-type kullanın. Desteklenen değerler: text, image_url.

• Çok sütunlu girişler için her sütunun içerik türünü belirtmek üzere content-types kullanın. Desteklenen değerler ve bunların Flink SQL sütun türleriyle eşleşmeleri aşağıda listelenmiştir:

• Parametrelerde content-types belirtmenizi öneririz. Hem content-type hem de content-types ayarlarsanız çerçeve hata bildirir. Not: Model parametrelerindeki ve çağrı başına ayarlardaki content-type ve content-types ayarları birbirini etkiler:

Sütun düzeyi parametreleri

Paket işleme

• İşleme: Paket işleme, önce birden çok giriş kaydını bir toplu işte birleştirir ve ardından bu toplu iş için yapay zeka modelini çağırır. Bu yaklaşım model çağrılarının sayısını azaltır ancak her istek için model çıkarım süresini artırır. Kayıtlar etkili bir şekilde birleştirilebildiğinde paket işleme, operatörün aktarım hızını önemli ölçüde artırabilir.

• Parametre açıklaması: bundle-size bir toplu işte biriktirilecek maksimum kayıt sayısını belirtir. bundle-allow-latency ise bir toplu iş bundle-size değerine ulaşmadığında izin verilen maksimum bekleme süresini belirtir. Toplu iş boyutu veya bekleme süresi koşullarından biri karşılandığında işleme başlar.

• Gereksinimler: Paket işleme, modelin giriş ve çıkış şemasına göre etki eder.

  • Metin oluşturma: task değeri chat/completions olan modeller için girişte tam olarak bir adet STRING türünde metin sütunu bulunur ve çıktıda tam olarak bir adet STRING veya VARIANT sütunu bulunur. Çok sütunlu girişler ve görüntü, video veya ses gibi çok modlu girişler desteklenmez.

  • Metin gömme: task değeri embeddings olan modeller için girişte tam olarak bir adet STRING sütunu bulunur ve çıktıda tam olarak bir adet ARRAY<FLOAT> sütunu bulunur.

• Desteklenen işlevler: ML_PREDICT, metin sınıflandırma (AI_CLASSIFY), duygu analizi (AI_SENTIMENT), bilgi çıkarımı (AI_EXTRACT), metin özetleme (AI_SUMMARIZE), metin çevirisi (AI_TRANSLATE), veri maskeleme (AI_MASK) ve metin gömme (AI_EMBED). Bunlardan ML_PREDICT yalnızca kayıtlı model yukarıdaki gereksinimleri karşıladığında paket işlemeyi destekler.

• Global yapılandırma: Bu yapılandırma, iş içindeki ML_PREDICT çağrılarına ve yukarıdaki gereksinimleri karşılayan listelenen alan odaklı yapay zeka işlevlerinin çağrılarına uygulanır. Paket işlemeyi etkinleştirmek için aşağıdaki global parametreleri kullanın:

  -- Bundle processing is supported only in VVR 11.9 and later.
  SET 'table.exec.ml-predict.bundle-size' = '10';
  SET 'table.exec.ml-predict.bundle-allow-latency' = '5s';

• Çağrı başına yapılandırma: Bu parametreler yalnızca belirtildikleri işlev çağrısına uygulanır. ML_PREDICT ve listelenen alan odaklı yapay zeka işlevleri aynı yapılandırma yöntemini kullanır. Parametreler ve örnekler için bkz. Çağrı başına yapılandırma.

Sınırlamalar

• Yalnızca VVR 11.9 ve sonraki sürümlerde desteklenir.

• Model çağrıları eşzamanlı modda çalışırken hata işleme etkinleştirilemez.

• Hata işleme şu anda Vektörleştirme sonucu önbelleği (CACHE_TABLE) ile birlikte kullanılamaz.

Hata işlemeyi etkinleştirme

Bir yapay zeka işlevini çağırırken FAIL_ON_ERROR => FALSE parametresini ekleyin.

Önkoşul: Modelin varsayılan yeniden deneme ayarlarını koruyun: error-handling-strategy='retry' ve retry-fallback-strategy='failover'. Model parametreleri hakkında daha fazla bilgi için bkz. Model ayarları.

Çıktı değişiklikleri

Hata işlemeyi etkinleştirdikten sonra sistem, orijinal çıkış alanlarının sonuna nullable VARIANT türünde bir err_msg hata sütunu ekler. Bu sütun sistem tarafından oluşturulur. Giriş tablosunda veya CREATE MODEL ifadesinin OUTPUT yan tümcesinde bildirmeniz gerekmez.

• Çağrı başarılı olursa: err_msg değeri NULL olur.

• Çağrı nihayetinde başarısız olursa: tüm model sonuç alanları NULL olur ve err_msg hata bilgilerini içerir.

SELECT * kullanırsanız sonuç yeni hata sütununu içerir. Alanları açıkça seçiyorsanız sütunu manuel olarak belirtmeniz gerekir. Varsayılan sütun adı err_msg şeklindedir. Ad mevcut bir çıkış sütunuyla çakışırsa sistem err_msg0 gibi sayısal bir son ek ekler. Sütuna başvurmak için gerçek sütun adını kullanın.

Örnekler

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)

Her hata türünün nasıl işlendiği

Hata işleme etkinleştirildiğinde ve varsayılan yeniden deneme ayarları korunduğunda aşağıdaki davranış geçerlidir. Hata satırı, girişi koruyan, boş model sonuçları döndüren ve başarısızlık nedenini err_msg alanında kaydeden bir satırdır.

Metin

Aşağıdaki örnekte, giriş metnini duyguya göre sınıflandırmak için Flink AI Service yerleşik bir modeli kaydedilir ve kullanılır. Örnek 1 MODEL düzeyi parametrelerini kullanır. Örnek 2 ise user-prompt değerini çağrı sırasında geçersiz kılar.

-- Register a built-in model. When content-type is not specified, it defaults to 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'
);

-- Create a source table: simulated product review data
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');

-- Create a result table
CREATE TEMPORARY TABLE output_table (
  id INT,
  content STRING,
  sentiment STRING
) WITH (
  'connector' = 'print'
);

-- Use ML_PREDICT for real-time inference
-- Example 1: Use MODEL-level parameters
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

-- Example 2: Specify per-call parameters
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']);

Örnek 1 çıktısı:

Örnek 2 çıktısı:

Görüntü

Aşağıdaki örnekte çok modlu bir model kaydedilir ve giriş görüntüleri sınıflandırılır.

Görüntü giriş sütunu (content-type değeri image_url olduğunda) STRING türündedir ve aşağıdaki formatları destekler:

Görüntüler bir Paimon Kataloğunun Blob sütunlarında depolanıyorsa, giriş olarak genel bir URL oluşturmak için Paimon Blob imzalı URL özelliğini kullanabilirsiniz. Daha fazla bilgi için bkz. Apache Paimon.

-- Register a built-in model with content-type set to 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'
);

-- Use ML_PREDICT for real-time inference
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

Video

Aşağıdaki örnekte, giriş videosunu anlamak için çok modlu bir model kaydedilir.

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));

Metin + tek görüntü

Aşağıdaki örnekte, metin ve görüntü girişleri üzerinde sohbet tabanlı çıkarım yapmak için çok modlu bir model kaydedilir.

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)
));         

Metin + birden çok görüntü (birden çok sütun)

Her görüntüyü ayrı bir INPUT sütunu üzerinden geçirin. content-types içindeki her görüntü sütunu için image_url belirtin.

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)
));          

Metin + birden çok görüntü (dizi)

Birden çok görüntüyü bir ARRAY<STRING> sütununda geçirin. content-types içinde multi_image_urls kullanın.

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)
)); 

Metin + 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));

-- When content-type / content-types are not specified in CREATE MODEL, the default is content-type = text
CREATE MODEL model_single
INPUT (input STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions'
);

-- Call 1: Use MODEL parameters, content type is text
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_text,
  MODEL model_single,
  DESCRIPTOR(input)
));

-- Call 2: Override content type to image at call time
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_img,
  MODEL model_single,
  DESCRIPTOR(input),
  MAP['content-type', 'image_url']
));

-- Set default configuration in CREATE MODEL (multi-column model)
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',
  'user-prompt' = 'Describe the image'
);

-- Call 1: Use default configuration
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_a, MODEL vl_model, DESCRIPTOR(text_input, image_input)
));

-- Call 2: Override user-prompt and column-level parameters
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_b, MODEL vl_model, DESCRIPTOR(text_input, image_input),
  MAP[
    'user-prompt', 'Answer in English',
    'image_input.min_pixels', '100',
    'image_input.max_pixels', '5000'
  ]
));

-- Call 3: Override content-types to change the type combination (treat both columns as text)
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_c, MODEL vl_model, DESCRIPTOR(text_input, image_input),
  MAP['content-types', 'text;text']
));

-- Call 4: Override content-types to change the type combination (reverse order)
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_c, MODEL vl_model, DESCRIPTOR(text_input, image_input),
  MAP['content-types', 'image_url;text']
));