ByteDance、ファイル即時アップロードによる多モーダルベクトル検索の実践を共有

オリジナル：Viking

2026年8月4日 19時12分 北京

企業が AI アプリケーションを概念実証（PoC）から本番環境へ移行する際、必ず遭遇するのが以下のようなシナリオです。商品画像のデータベースには毎日数千枚の新規写真が追加され、企業向け知識ベースには新たなドキュメントが絶えず流入し、訓練データプラットフォームでは新しいサンプルを即座に検知する必要があります。こうしたデータの最初の保管先は、通常オブジェクトストレージとなります。

しかし、「ファイルをアップロードした」ことと「AI がその内容を活用できる」ことは別問題です。「保存する」状態から「利用する」状態へ移行するには、一般的に以下のフローを経由します。

新規または変更されたオブジェクトの検出 → オブジェクトまたはメタデータの読み取り → モデル入力のクリーニングと組み立て → Embedding（ベクトル化）の生成 → ベクトルデータベースへの書き込み → 検索インデックスの更新

もし上記のプロセスが複数の定期タスクやスクリプトを繋ぎ合わせたものであれば、本番環境では以下の3 つのボトルネックに直面することが多くなります。

データ更新の遅延：新ファイルは次のスキャンサイクルまで待たされるため、検索対象となる情報が数時間、あるいはそれ以上遅れる可能性があります。

既存データと新規データの連携困難：全量スキャン中に新たなファイルがアップロードされると、インクリメンタル（増分）処理への切り替え時に、データの抜けや重複が発生しやすくなります。

モデルとベクトル書き込みの複雑なリンク構築：画像の取得、モデルサービスの提供、GPU リソースの確保、失敗時の再試行、ベクトルの書き込み、インデックスの更新など、それぞれを個別に実装する必要があります。

これらの課題を解決するため、本稿では火山引擎（Volcengine）の Flink と VikingDB を組み合わせて、「ファイルアップロード後、数秒で検索可能になる」リアルタイムなマルチモーダルデータリンクを構築する方法を紹介し、2 つの完全な Flink SQL 参考実装案を提示します。

一、一つのリンクですべての工程を統合

Flink と VikingDB の連携ソリューションは、上記のように分散していた工程を、継続して実行されるリアルタイムデータリンクへと集約します。

このデータ処理パイプラインを構成する主要なコンポーネントは以下の通りです。

TOS（オブジェクトストレージ）: 画像、動画、テキスト、ドキュメント、および業務メタデータを格納します。多模态データの保管場所となります。

Kafka（メッセージキュー）: TOS での PUT や DELETE などのオブジェクトイベントを処理します。ファイルに変更が生じると、自動的に Kafka にメッセージが送信されます。

Flink（ストリーム計算エンジン）: 全量および増分データの取り込み、データクリーニング、ルーティング、モデル呼び出し、そして障害復旧を担当します。このパイプライン全体を制御する「オーケストレーション・エンジン」です。

VikingDB（ベクトルデータベース）: データのベクトル化、保存、インデックス作成、およびオンライン検索を担います。ベクトルデータの最終的な保管先であり、検索サービスのバックエンドとなります。

方舟（Ark）: 標準モデルでは対応できないカスタム要件を持つシーン向けに、多模态な埋め込み（Embedding）機能を提供します。組み込みモデルが不十分な場合に、独自のモデルをここに接続して利用できます。

2. 3 つの主要機能でリアルタイム AI データパイプラインを実現

1. TOS-CDC：オブジェクトストレージを継続的に更新されるテーブルへ変換

TOS-CDC は、オブジェクトストレージ向けの Flink SQL Source Connector です。もともと散在していたオブジェクトストレージ内のファイル変更を、連続的に処理可能なデータストリームに変換します。

ジョブ起動後、以下の動作を行います：

• 全量スキャンの開始時間を記録する
• 指定された TOS バケットに対して全量スキャンを実行する
• 全量スキャン完了後は、デフォルトでその開始時間に対応する Kafka の位置からオブジェクトイベントを消費し始める
• チェックポイントにより、全量スキャンの進捗状況と Kafka の消費位点を保存する

全量スキャン時点から Kafka 上の増分データを継続して消費することで、スキャン期間中に発生したオブジェクトの変更もすべてカバーできます。下流システムで安定した主鍵を持つ Upsert（「存在すれば更新し、なければ挿入」）処理を行うことで、重複イベントが発生しても最終的には単一のレコードに収束します。

注：TOS-CDC が受け取るのはファイルの場所に関する情報であり、ファイルそのもののコンテンツは受け取りません。

1. VikingDB：書き込み・ベクトル化・検索を一つのループで完結させる

標準的な画像やテキストの検索シナリオでは、VikingDB のテーブル上でフィールドの意味とベクトルモデルを宣言するだけで済みます。例えば、フィールドを image として宣言し、doubao-embedding-vision を設定すれば、VikingDB が自動的に画像の読み込み、ベクトル化、およびインデックス更新を担当します。

業務側で追加にモデルサービスの維持管理や GPU リソースプールの用意、ベクトル取り込みプログラムの構築は不要です。Flink が変化するデータを継続的に書き込み、VikingDB がそれを検索可能なベクトル資産として蓄積・定着させます。

VikingDB Connector は Flink の Changelog に基づいて Upsert と Delete を実行できます。デフォルトでは同期書き込みが設定されています：

'async' = 'false'

書き込み直後に高速な検索が必要な業務の場合、async=true を直接有効化することは推奨されません。非同期書き込みはスループット優先の設計であり、Collection（データ集合）と Index の可視性に遅延が生じる可能性があります。

1. Flink 2.2 AI SQL：モデル呼び出しを SQL に組み込む

一部の業務では、Flink 側で独自に Embedding を処理する必要があります。典型的なシナリオは以下の通りです：

• 画像、タイトル、タグ、OCR テキストを組み合わせて多モーダル入力を作成する
• 指定された方舟モデルとそのバージョンを使用する
• モデル呼び出しの並列度、スループット、コストを一元的に制御する
• 一つの Embedding を VikingDB、Kafka、特徴量ストア、または訓練サンプルへ同時に書き込む
• Java や Python のジョブを再実装することなくモデルを切り替える

ストリーミング計算プラットフォーム Flink バージョン 2.2 では、CREATE MODEL で方舟モデルを宣言し、ML_PREDICT を SQL 内で使用することでリアルタイム推論を実行できます。モデルの接続、データ処理、ベクトル書き込みが一つの SQL ジョブで統括的にオーケストレーションされます。

注：現在の Flink バージョン 2.2 の AI SQL は招待制テスト中です。ご要望の場合は火山公式サイトまでお問い合わせください。

三．事前準備チェックリスト

リンク構築を開始する前に、以下のリソースを準備してください：

• 準備項目：役割 / 必須か
• TOS Bucket とイベント通知ルール：多モーダルオブジェクトの保存と、オブジェクト変更イベントの Kafka への転送 / 必須
• Kafka インスタンス、トピック、ユーザー：TOS の増分イベントを受け取り、TOS-CDC が継続的に消費する / 必須
• ストリーミング計算 Flink バージョン：TOS-CDC、データ処理、VikingDB Sink の実行環境 / 必須
• TOS-CDC 招待制テスト資格：全量スキャンと増分イベントを一体化した Source を利用可能にする / 必須
• Flink 2.2 招待制テスト資格：CREATE MODEL と ML_PREDICT を使用して方舟モデルを呼び出す / 方案二のみ必要
• VikingDB インスタンスと API Key：自動ベクトル化の実行、または Flink が生成したベクトルの保存 / 必須
• 方舟推論エンドポイントと API Key：カスタム多モーダル Embedding の実行 / 方案二のみ必要

注：両方のアプローチについては後ほど詳しく解説します。

ファイルアップロードで即座に検索可能｜リアルタイム多モーダルベクトルリンクの実践

1. TOS バケットの準備とイベント通知の有効化

TOS のイベント通知機能は、バケット内のオブジェクトに変更が発生した際に、その事象を Kafka へプッシュします。このイベントメッセージには、バケット名、オブジェクトキー、イベントの種類、発生時刻などの情報が含まれており、TOS-CDC はこれらを常時監視することで、新規作成、上書き、削除の各操作を検知します。

TOS コンソールで対象バケットにイベント通知ルールを作成する際は、以下の設定が必要です。

• 配信先を「メッセージキュー Kafka 版」に指定します。
• 少なくとも tos:ObjectCreated:* を購読してください。オブジェクトの削除処理も必要になった場合は、後から tos:ObjectRemoved:* も追加で購読します。
• バッファ範囲に合わせて Prefix と Suffix を設定し、イベントの対象範囲を TOS-CDC の bucket 設定と整合させます。
• 対象となる Kafka インスタンス、Topic、Kafka ユーザー、および権限付与ロールを選択します。

注：詳細な手順については、火山エンジンオブジェクトストレージのドキュメント「イベント通知を Kafka へプッシュする方法」をご参照ください。（リンク先：https://docs.volcengine.com/docs/6349/1817509?lang=zh）

1. Kafka インスタンス、Topic、およびアクセス権限の準備

メッセージキュー Kafka 版において、インスタンス、Topic、およびアクセスユーザーを作成します。TOS のイベント通知ルールでは、Kafka インスタンス ID、Topic 名、ユーザー情報、IAM ロールを参照する必要があります。このロールには、TOS から Kafka へのイベント配信を許可するためのシステム事前定義ポリシー KafkaAccessForTOS をバインドしておく必要があります。

併せて、以下の点も確認してください。

• Flink リソースプールから Kafka Bootstrap Servers へアクセス可能であること。
• Kafka ユーザーと認証パラメータが Flink ジョブ内で使用可能であること。
• Topic の Retention（保持期間）を「全量スキャンの最大所要時間 + rewind.offset」よりも長く設定し、全量スキャン完了後に遡って処理が必要なイベントが期限切れになるのを防ぎます。

1. 流式計算 Flink 版の有効化

火山エンジン流式計算 Flink 版を有効化し、プロジェクトとジョブ実行に必要なリソースプールを作成します。さらに、Kafka、VikingDB、および方舟サービスとのネットワーク接続を確立してください。

注：TOS-CDC のベータテスト参加資格や Flink 2.2 バージョンのベータテスト参加資格が必要な場合は、火山エンジンの公式サイトまでお問い合わせください。

1. VikingDB と方舟リソースの準備

VikingDB を有効化し、データ面アドレスと API Key を用意します。方案一では、対象 Collection で使用する自動ベクトル化モデルの種類、バージョン、次元数を確認する必要があります。方案二では、方舟の推論アクセスポイントを作成し、モデル名、出力次元、API Key を準備します。

注：すべてのアクセス認証情報は、流式計算 Flink 版の暗号化変数または実行環境変数を通じて注入することを推奨します。SQL 内に直接記述するのは避けてください。

三、2 つの SQL 方案：用途に応じて選んでください

両方の方案は、同じ TOS-CDC データ取り込みリンクを共有しますが、「どちらがベクトル化処理を担当するか」で異なります。

• 方案一：VikingDB の自動ベクトル化。標準的な画像・テキスト検索向け。最もシンプルなアーキテクチャです。データを VikingDB に渡せば、同社が自動的に Embedding を生成します。
• 方案二：Flink AI SQL で方舟を呼び出し。画像とテキストの融合処理、モデルの自主制御、ベクトルの多様な下流利用向け。ユーザーがモデル、入力、出力を完全にコントロールできます。

1. TOS-CDC ソーステーブルの作成

CREATE TABLE tos_object_events (

object_key STRING NOT NULL,

object_url STRING,

bucket_name STRING,

file_name STRING,

object_etag STRING,

object_size BIGINT,

mtime TIMESTAMP_LTZ(3),

event_time TIMESTAMP_LTZ(3),

record_origin STRING,

PRIMARY KEY (object_key) NOT ENFORCED

) WITH (

'connector' = 'tos-cdc',

'path' = 'tos://my-bucket/images01/, tos://my-bucket/images02/',

'properties.bootstrap.servers' = 'kafka.example:9092',

'properties.group.id' = 'ingest-cg',

'topic' = 'object-events',

'scan.startup.mode' = 'initial'

);

本番環境では Kafka インスタンスに応じて認証情報やネットワークパラメータを追加する必要があります。また、以下の点に注意してください。

Kafka Topic のメッセージ保持期間は、全量スキャンから Kafka への切り替え完了まで、および設定された rewind.offset をカバーできる時間以上に設定してください。

増分コールバックのパラメータ rewind.offset はデフォルトで 0 ですが、必要に応じて変更可能です。また、位置が期限切れになった際にデータが静かに欠落するのを防ぐため、rewind.retention-miss-policy=fail を維持することを強く推奨します。

出力データの構築：

CREATE TEMPORARY VIEW image_put_events AS

SELECT

object_key AS id,

object_url AS image_uri,

object_etag,

COALESCE(event_time, mtime) AS update_time

FROM tos_object_events;

ここでは object_key を主鍵として使用しています。TOS-CDC と外部 Sink の双方が「少なくとも 1 回」のセマンティクス（At-Least-Once）を採用しているため、故障復旧時や全量・増分の接続時にレコードが重複して再プレイされる可能性があります。安定した主鍵を設定することで、VikingDB 内で重複する PUT 操作が Upsert として処理され、最終的に単一のデータに収束します。

1. 案一：VikingDB の自動ベクトル化

まず VikingDB Catalog を作成します：

CREATE CATALOG viking WITH (

'type' = 'vikingdb',

'control-plane.host' = 'open.volcengineapi.com',

'region' = 'cn-beijing',

'project-name' = 'default',

'access-key' = '${secret_values.volc-ak}',

'secret-key' = '${secret_values.volc-sk}',

'data-plane.host' = '',

'api-key' = '${secret_values.vikingdb-api-key}'

);

次に、自動画像ベクトル化を有効にした Collection を作成します。

CREATE TABLE IF NOT EXISTS viking.default.realtime_image_assets (

id STRING,

image_uri STRING,

object_etag STRING,

update_time TIMESTAMP_LTZ(3),

PRIMARY KEY (id) NOT ENFORCED

) WITH (

'vikingdb.field.image_uri.type' = 'image',

'vikingdb.vectorize.dense.model-name' = 'doubao-embedding-vision',

'vikingdb.vectorize.dense.model-version' = '',

設定値として、vikingdb.vectorize.dense.dim に 2048 を、vikingdb.vectorize.dense.image-field に image_uri を指定します。

次に VikingDB への書き込みを行います。

INSERT INTO `viking`.`default`.`realtime_image_assets`
SELECT id, image_uri, object_etag, update_time
FROM image_put_events;

この処理フローで実現できるのは、ジョブ初回起動時に指定したストレージバケット内の既存オブジェクトを一括インポートすることです。 （原文の技術表記: viking、default、realtime_image_assets）

新对象上传後はリアルタイムで書き込みを行います。

同じパスのオブジェクトが上書きされた場合は、安定した主キーに基づいて更新します。

ジョブ失敗時にはチェックポイントから復元し、Upsert によってイベントの再実行を防ぎます。

1. 案二：Flink AI SQL を用いた方舟 Embedding の呼び出し

業務でカスタム多模態入力が必要、または特定のモデルを指定する必要がある場合、同一の TOS-CDC ソーステーブルを再利用できます。

まず SQL で方舟モデルを宣言します：

CREATE MODEL ark_multimodal_embedding

INPUT (payload STRING)

OUTPUT (embedding ARRAY)

WITH (

'provider' = 'ark',

'endpoint' = 'https://ark.cn-beijing.volces.com/api/v3/embeddings/multimodal',

'api-key' = '${secret_values.ark-api-key}',

'model' = 'doubao-embedding-vision-251215',

'model.dimensions'= '2048'

);

オブジェクト情報をアーク（Ark）の多モーダル入力形式に組み立てます：

CREATE TEMPORARY VIEW multimodal_payload AS

SELECT

id,

image_uri,

update_time,

CAST(

JSON_ARRAY(

JSON_OBJECT(

'type' VALUE 'image_url',

'image_url' VALUE JSON_OBJECT(

'url' VALUE CONCAT(

'https:///',

object_key

)

)

)

) AS STRING

) AS payload

FROM (

SELECT

object_key AS id,

object_url AS image_uri,

COALESCE(event_time, mtime) AS update_time,

object_key

FROM tos_object_events

) AS source_events;

例として、HTTPS でアクセス可能な画像 URL をモデルの入力に使用しています。本番環境では、アークサービスがそのアドレスを安全にアクセスできることを保証する必要があります。プライベートなバケットを使用する場合は、制御された一時署名付き URL や社内認証リンクを利用し、モデル呼び出しのためにバケット全体を公開読み取り用に設定することは推奨されません。

明示的なベクトルを保存するための VikingDB Collection を作成します：

CREATE TABLE IF NOT EXISTS viking.default.realtime_multimodal_assets (

id STRING,

image_uri STRING,

update_time TIMESTAMP_LTZ(3),

mixed_embedding ARRAY,

PRIMARY KEY (id) NOT ENFORCED

) WITH (

'vikingdb.field.mixed_embedding.type'= 'vector',

'vikingdb.field.mixed_embedding.dim' = '2048'

);

モデルを呼び出して VikingDB に書き込みます：

INSERT INTO viking.default.realtime_multimodal_assets

SELECT

id,

image_uri,

update_time,

embedding AS mixed_embedding

FROM ML_PREDICT(

TABLE multimodal_payload,

MODEL ark_multimodal_embedding,

DESCRIPTOR(payload)

);

もし Embedding をリアルタイム特徴量や訓練サンプル、メッセージ購読に利用する必要がある場合、EXECUTE STATEMENT SET を使用して複数の Sink を追加し、下流システムで同じモデル計算結果を共有できるようにします。

五、このルートの検証方法

1. Flink タスクが実行状態にあるか確認

Flink UI を通じて、既存の画像や動画ファイルが VikingDB に正常にインポートされていることを確認してください。データ量と TOS の整合性が取れているかも併せてチェックします。

2. ベクトルと検索結果の検証

VikingDB コンソールの「データセット → データプレビュー」で、TOS のパスに沿ってクエリを実行し、データが正しくデータセットに書き込まれているか確認します。

VikingDB の Collection 内のフィールドタイプやベクトル次元数を確認した上で、類似画像や関連テキストを用いて検索を実行し、直近でアップロードされたオブジェクトが正しく召回（検索結果にヒット）されるか検証します。以下の図のように、「小松鼠」と入力すると、関連する類似写真が検索結果として表示されます。

3. エンドツーエンドの遅延測定

以下の項目をそれぞれ記録します。

• TOS オブジェクトのアップロード時間
• Kafka イベント発生時刻
• Flink の処理時間
• VikingDB への書き込みが可視化されるまでの時間
• そのオブジェクトを初めて検索できるまでの時間

P50 および P95 の遅延を、実際のデータ規模と並行処理条件に基づいて評価した上で、Flink の並行数、モデルのスループット、Sink Flush Interval、そして VikingDB のインデックス設定を調整してください。

1. VikingDB を活用した多様な検索テストの実施

データをリアルタイムで VikingDB に書き込んだ後、ビジネスのシナリオに応じて異なる検索方式を選択できます：

• 検索方式：適用シーン / 詳細ドキュメント
• 多模態検索：画像からの類似画像検索、テキストによる画像検索、画像とテキストを混合した検索。画像・テキスト・混合入力をサポートします。 / https://docs.volcengine.com/docs/84313/1791135?lang=zh
• キーワード検索：完全一致や用語・番号類のクエリに最適。全文検索と意味検索を補完し合います。 / https://docs.volcengine.com/docs/84313/1791139
• 地理情報検索：地理位置および距離に基づいた検索フィルタリングをサポートします。 / https://docs.volcengine.com/docs/84313/1791133?lang=zh#filter%E7%BB%93%E6%9E%84

最後に

多模態 AI アプリケーションが本番環境で運用されるようになると、その価値はモデルの性能だけでなく、データの更新速度にも左右されます。画像、動画、音声、ドキュメントといった非構造化データにおいても、オブジェクトの内容やメタデータを Flink に接続できれば、「イベントトリガー型」「全量と増分を一体化した」「書き込み即検索可能」というアプローチで、企業向け AI アプリケーションのリアルタイムベクトル化パイプラインを構築することが可能です。

このパイプラインがもたらす核心的な変化は以下の通りです：

• イベント駆動によるスキャンの置換: ファイルアップロード後、数秒以内に処理がトリガーされます。データの可視化遅延は時間単位から数秒レベルに短縮されました。
• 単一の SQL ジョブで全ルートを貫通: 全量データ、増分データ、ベクトル化、そしてストレージを一つの SQL で完結させることで、アーキテクチャの複雑さが劇的に低下しました。
• ベクトル化能力は即戦力かつ自律可能: VikingDB の内蔵モデルを使えばゼロからエンジニアリングする必要もありませんし、Flink 2.2 AI SQL を用いて方舟（Ark）を呼び出すことで、モデルの自主的な運用も可能です。
• 書き込み後数秒で検索可能: 生成された Collection はそのまま、ナレッジベースの質問応答、推薦時の召回、そして多模態検索を支えることができます。

これはつまり、企業ナレッジベースの更新スピードが向上し、コンテンツ推薦が新しい素材を即座に感知できるようになり、トレーニング用サンプルもより迅速に蓄積されることを意味します。

火山引擎（Volcengine）の Flink と VikingDB を組み合わせることで、「多様なソース」「多様なモダリティ」「絶えず変化する」データをリアルタイムで検索可能かつサービス可能なベクトル資産へと変換。これにより、企業の AI アプリケーションは PoC（概念実証）から着実に本番環境へと移行していくことができます。

WeChat で開くにはこちらへ