AI_MASK で Realtime Compute for Apache Flink SQL の機密データをマスキングして PII を保護 - Flink

このトピックでは、AI_MASK を使用して大規模言語モデル（LLM）によるデータマスキングを実行する方法について説明します。

制限事項

Ververica Runtime (VVR) 11.4 以降でのみサポートされます。
AI_MASK 文で使用される Flink 演算子のスループットは、モデルサービスプラットフォームにおけるレート制限により制御されます。トラフィックがプラットフォームのアクセス制限に達すると、Flink ジョブ内でバックプレッシャーが発生し、AI_MASK 演算子がボトルネックとなります。深刻な場合は、関連する演算子がタイムアウトしてジョブが再起動される場合があります。

構文

AI_MASK(
  MODEL => MODEL <MODEL NAME>, 
  INPUT => <INPUT COLUMN NAME>,
  MASK_ENTITIES => <MASK ENTITIES>
)

入力パラメーター

パラメーター	データ型	説明
MODEL <MODEL NAME>	MODEL	登録済みのモデルサービスの名前です。詳細については、「モデル設定」をご参照ください。注：モデルは VARIANT 型の出力を返す必要があります。
<INPUT COLUMN NAME>	STRING	モデルが分析する生テキストです。
<MASK ENTITIES>	ARRAY<STRING>	マスク対象のエンティティです。注：このパラメーターは定数である必要があります。

出力

パラメーター	データ型	説明
masked_text	STRING	マスク済みテキストです。
detected_entities	ARRAY<STRING>	検出されたエンティティです。

例

テストデータ

id	content
1	ティモは本当に勉強が大好きです。彼は空き時間があると、いつでも学習資料を読みます。

テスト文

以下の SQL 文は、Qwen-Plus モデルを作成し、AI_MASK を使用してデータをマスクします。

CREATE TEMPORARY MODEL general_model
INPUT (`input` STRING)
OUTPUT (`content` VARIANT)
WITH (
    'provider' = 'openai-compat',
    'endpoint'='<YOUR ENDPOINT>',
    'apiKey' = '<YOUR KEY>',
    'model' = 'qwen-plus'
);

CREATE TEMPORARY VIEW infos(id, content)
AS VALUES (1, '田中一郎は学習が大好きです。暇な時間には常に学習資料を読みます。');

-- 位置引数を使用して AI_MASK を呼び出す
SELECT id, masked_text, detected_entities
FROM infos,
LATERAL TABLE(
  AI_MASK(
    MODEL general_model, 
    content, 
    ARRAY['name']
    ));

-- 名前付き引数を使用して AI_MASK を呼び出す
SELECT id, masked_text, detected_entities
FROM infos,
LATERAL TABLE(
  AI_MASK(
    MODEL => MODEL general_model, 
    INPUT => content, 
    MASK_ENTITIES => ARRAY['name']
    ));

出力結果

ID	マスク済みテキスト	検出されたエンティティ
1	[NAME] は本当に勉強が好きです。空き時間があればいつでも学習資料を読みます。	[{"entity":"Timmo","type":"name"}]