Realtime Compute for Apache Flink:汎用呼び出し

コンテンツタイプパラメーター

• 単一列入力の場合、content-type を使用してコンテンツタイプを指定します。サポートされている値は、text、image_url です。

• 複数列入力の場合、content-types を使用して各列のコンテンツタイプを指定します。サポートされている値と Flink SQL の列タイプへのマッピングは以下のとおりです。

• パラメーターでは content-types を指定することを推奨します。 content-type と content-types の両方を設定すると、フレームワークがエラーを返します。 注： モデルパラメーターと呼び出しごとの設定では、content-type と content-types の設定が互いに影響します。

列レベルのパラメーター

バンドル処理

• 処理： バンドル処理は、まず複数の入力レコードをバッチに集約し、そのバッチに対して AI モデルを呼び出します。これにより、モデルの呼び出し回数は減少しますが、各リクエストのモデル推論時間は増加します。レコードを効果的に集約できる場合、バンドル処理は演算子のスループットを大幅に向上させることができます。

• パラメーターの説明： bundle-size はバッチに蓄積されるレコードの最大数を指定します。bundle-allow-latency は、バッチが bundle-size の値に達していない場合の最大待機時間を指定します。バッチサイズまたは待機時間のいずれかの条件が満たされると、処理が開始されます。

• 要件： バンドル処理は、モデルの入出力スキーマに基づいて有効になります。

  • テキスト生成： task が chat/completions であるモデルの場合、入力にはテキストの STRING 列が 1 つだけ含まれ、出力には STRING または VARIANT 列が 1 つだけ含まれます。複数列入力や、画像、動画、音声などのマルチモーダル入力はサポートされていません。

  • テキスト埋め込み： task が embeddings であるモデルの場合、入力には STRING 列が 1 つだけ含まれ、出力には ARRAY<浮動小数点> 列が 1 つだけ含まれます。

• サポートされる関数：サポートされる関数は、ML_PREDICT、テキスト分類 (AI_CLASSIFY)、感情分析 (AI_SENTIMENT)、情報抽出 (AI_EXTRACT)、テキスト要約 (AI_SUMMARIZE)、テキスト翻訳 (AI_TRANSLATE)、データマスキング (AI_MASK)、およびテキスト埋め込み (AI_EMBED) です。このうち、ML_PREDICT は、登録されたモデルが前述の要件を満たす場合にのみバンドル処理をサポートします。

• グローバル設定： この設定は、前述の要件を満たすジョブ内の ML_PREDICT 呼び出しおよび記載のドメイン固有 AI 関数の呼び出しに適用されます。バンドル処理を有効にするには、次のグローバルパラメーターを使用します。

  -- バンドル処理は VVR 11.9 以降でのみサポートされています。
  SET 'table.exec.ml-predict.bundle-size' = '10';
  SET 'table.exec.ml-predict.bundle-allow-latency' = '5s';

• 呼び出しごとの設定： これらのパラメーターは、パラメーターが指定された関数呼び出しにのみ適用されます。ML_PREDICT および記載のドメイン固有 AI 関数は、同じ設定方法を使用します。パラメーターと例については、「呼び出しごとの設定」をご参照ください。

制限

• VVR 11.9 以降でのみサポートされています。

• モデルの呼び出しが同期モードで実行される場合、エラーハンドリングを有効にすることはできません。

• 現時点では、エラー処理はベクトル化結果キャッシュ (CACHE_TABLE) と併用できません。

エラーハンドリングの有効化

AI 関数を呼び出す際に、FAIL_ON_ERROR => FALSE パラメーターを追加します。

前提条件：モデルのデフォルトの再試行設定である error-handling-strategy='retry' および retry-fallback-strategy='failover' を維持します。モデルパラメーターの詳細については、「モデル設定」をご参照ください。

出力の変更

エラー処理を有効にすると、システムは NULL 値を許容する VARIANT 型のエラー列 err_msg を元の出力フィールドの末尾に追加します。この列はシステムによって生成されるため、入力テーブルまたは CREATE MODEL の OUTPUT 句で宣言する必要はありません。

• 呼び出し成功: err_msg は NULL です。

• 最終的に呼び出しは失敗し、すべてのモデル結果フィールドが NULL になり、err_msg にエラー情報が含まれます。

SELECT * を使用すると、結果に新しいエラー列が含まれます。 フィールドを明示的に選択する場合は、列を手動で指定する必要があります。 デフォルトの列名は err_msg です。 名前が既存の出力列と競合する場合、システムは err_msg0 などの数値サフィックスを追加します。 列を参照するには、実際の列名を使用してください。

例

CREATE TEMPORARY VIEW input_data AS
SELECT *
FROM (VALUES
  (1, CAST('xxxx (人間性に反する暴力的な紛争の記述)' AS STRING), 'negative'),
  (2, CAST('この店のカスタマーサービスは常に最高レベルです' 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)

各エラータイプの処理方法

エラー処理が有効で、デフォルトの再試行設定が維持されている場合、次の動作が適用されます。エラー行とは、入力を保持し、空のモデル結果を返し、失敗の理由を err_msg に記録する行のことです。

テキスト

次の例では、Flink AI サービスの組み込みモデルを登録して使用し、入力テキストを感情で分類します。例 1 では MODEL レベルのパラメーターを使用します。例 2 では、呼び出し時に user-prompt を上書きします。

-- 組み込みモデルを登録します。content-type を指定しない場合、デフォルトは 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 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 TEMPORARY TABLE output_table (
  id INT,
  content STRING,
  sentiment STRING
) WITH (
  'connector' = 'print'
);

-- ML_PREDICT を使用してリアルタイム推論を実行します
-- 例 1：MODEL レベルのパラメーターを使用します
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

-- 例 2：呼び出しごとのパラメーターを指定します
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']);

例 1 の出力：

例 2 の出力：

画像

次の例では、マルチモーダルモデルを登録し、入力画像を分類します。

画像入力カラム (content-type が image_url の場合) は STRING 型で、次の形式をサポートします。

画像が Paimon カタログの BLOB 列に保存されている場合は、Paimon Blob の署名付き URL 機能を使用して、入力として使用するパブリック URL を生成できます。詳細については、「Apache Paimon」をご参照ください。

-- content-types を image_url に設定してマルチモーダルモデルを登録します
CREATE TEMPORARY MODEL image_describer_model
INPUT (image_url STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.5-plus',
  'content-types' = 'image_url',
  'system-prompt' = 'Describe the image briefly.'
);

-- ソースビューを作成します
CREATE TEMPORARY VIEW image_source(url) AS VALUES ('https://www.example.com/cat.jpg');

-- ML_PREDICT を使用して推論を実行します
SELECT
  url,
  response
FROM ML_PREDICT(
  TABLE image_source,
  MODEL image_describer_model,
  DESCRIPTOR(url));

動画

次の例では、入力動画を理解するためにマルチモーダルモデルを登録します。

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

テキスト + 単一画像

次の例では、テキストと画像入力に対するチャットベースの推論に使用するマルチモーダルモデルを登録します。

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

テキスト + 複数画像 (複数カラム)

各画像を個別の INPUT カラムで渡します。content-types で、各画像カラムに image_url を指定します。

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

テキスト + 複数画像 (配列)

ARRAY<STRING> カラムで複数の画像を渡します。content-types で multi_image_urls を使用します。

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

テキスト + 動画

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