はじめに：Kafka で非構造化データを扱う難しさ
Kafka はもともと構造化されたイベントデータを高速・大量に流すために設計されたメッセージバスです。ログ、トランザクション、センサーデータなど、スキーマが定義された比較的小さなメッセージを扱うのが得意です。
一方で、PDF・画像・動画といった非構造化データをパイプラインに乗せようとすると途端に話が複雑になります。Kafka のメッセージサイズには上限があり、大きなバイナリをそのままトピックに流すのは現実的ではありません。
この記事では、Confluent（Kafka）で非構造化データを扱う方法として Claim Check パターンを紹介し、PDF をアップロードしてベクトル DB に格納するまでのパイプラインの実装例を紹介します。
2つのアプローチ：Event Chunking と Claim Check
非構造化データを Kafka パイプラインに乗せる方法は、大きく2つに分類できます。
アプローチ① Event Chunking
大きなファイルを複数の小さなチャンクに分割して、それぞれを個別の Kafka メッセージとして送る方式です。
メリット
- Kafka だけで完結する（外部ストレージ不要）
- ストリーミング処理がしやすい
デメリット
- Consumer 側でチャンクを再結合する実装が必要
- チャンクの順序保証・欠損検知が複雑
- Kafka ブローカーへの負荷が高い
- RAG のようにファイル単位で意味的な処理をしたい場合に不向き
アプローチ② Claim Check
ファイル本体は外部ストレージ（S3 など）に保存し、Kafka には「このファイルがここにある」という参照情報（URI）だけを流す方式です。ホテルのクロークで荷物を預けてチケットだけ持ち歩くイメージです。
メリット
- Kafka に流れるメッセージは軽量（URI のみ）
- ファイル本体の処理は Consumer が S3 から取得して自由に行える
- 大きなファイルでも Kafka の設定変更不要
- S3 側で独立したライフサイクル管理（バージョニング・バックアップ）が可能
デメリット
- 外部ストレージへの依存が増える
- Consumer が S3 にアクセスできる必要がある
どちらを選ぶべきか
| 観点 | Event Chunking | Claim Check |
|---|---|---|
| Kafka のみで完結 | ✅ | ❌ |
| 大きなファイル（1MB超） | ❌ Kafkaに高負荷 | ✅ |
| 再結合ロジック | ❌ 複雑 | ✅ 不要 |
| RAG など意味的処理 | ❌ 全チャンク再結合後に処理が必要 | ✅ S3 から完全なファイルをそのまま取得できる |
| 本番運用のシンプルさ | ❌ | ✅ |
PDF・画像など意味的なまとまりを持つ非構造化データの場合は Claim Check 方式が実装しやすく、実用的かと思い、試しにClaimCheck方式でパイプラインを作ってみました。
実装：Claim Check 方式で作る RAG パイプライン
Claim Check 方式を採用し、PDF をアップロードすると自動的にテキスト抽出・Embedding が行われ、ベクトル DB に格納されるパイプラインを構築しました。なお、ベクトル DB に格納されたデータを使った RAG クエリ（ベクトル検索 + LLM による回答生成）の部分はこの記事の対象外です。
使用サービス
| 役割 | サービス |
|---|---|
| メッセージバス | Confluent Cloud (Kafka) |
| ストリーミング処理 + Embedding | Confluent Cloud Flink SQL |
| ファイルストレージ（Claim） | AWS S3 |
| Embedding モデル | AWS Bedrock (Titan Embed v2) |
| PDF テキスト抽出 | Docling（IBM OSS） |
| ベクトル DB | AstraDB (DataStax) |
| UI | Streamlit |
事前準備
パイプラインを動かすには以下のサービスのアカウントと認証情報が必要です。取得した値は .env ファイルに設定してください。
Confluent Cloud
Confluent Cloud のアカウントを作成し、以下を用意します。
| 必要なもの | 取得場所 |
|---|---|
| Kafka クラスター（Bootstrap サーバー） | Confluent Cloud コンソール → Cluster → Cluster Settings |
| API キー / シークレット | Confluent Cloud コンソール → API Keys |
| Schema Registry URL / API キー | Confluent Cloud コンソール → API Keys |
セットアップ手順は以下の記事が参考になります。
📄 Confluent Cloudで始めるApache Kafka & Flink入門（Zenn）
📄 Confluent Cloud API Key の Webコンソールでの作成方法（Qiita）
AWS S3 & Bedrock
AWS アカウントで以下を用意します。
| 必要なもの | 備考 |
|---|---|
| S3 バケット | PDF の保存先。パブリックアクセスはブロックしたまま使用可 |
| IAM ユーザーの Access Key / Secret | S3 への読み書きと Bedrock の呼び出し権限が必要 |
| Bedrock モデルアクセス |
amazon.titan-embed-text-v2:0 を有効化（AWS コンソール → Amazon Bedrock → Model access） |
AstraDB（DataStax）
Embedding（1024 次元ベクトル）を格納するコレクションを事前に作成しておきます。作成手順は以下の記事を参照してください。
コレクション作成時は以下の設定にしてください。
| 設定項目 | 値 |
|---|---|
| ベクトル次元数 | 1024 |
| 類似度関数 | Cosine |
作成後、API エンドポイント・トークン・キースペース・コレクション名を .env に設定してください。
アーキテクチャ全体図
パイプラインは5つのフェーズで構成されています。① Upload でユーザーが PDF をアップロードすると、② Claim Check でファイル本体は S3 に保存され、Kafka には軽量な URI イベントだけが流れます。③ PDF 解析 フェーズでは Extractor が S3 から PDF を取得し、Docling でテキスト抽出・チャンク化します。④ Embedding では Confluent Cloud Flink の AI_EMBEDDING() 関数が SQL だけでチャンクをリアルタイムにベクトル化します。⑤ 書き込み で AstraDB にインデックスされ、ベクトル検索に使えるデータとして格納されます。
コンポーネント構成
このパイプラインは 4つの独立したプロセス で構成されています。それぞれ別々に起動し、Kafka トピックを介してデータを受け渡します。
rag-pipeline/
├── uploader/
│ └── app.py # Streamlit UI。PDF アップロード時に S3 + Kafka への送信を実行
├── extractor/
│ └── app.py # file-uris を常時 consume して Docling でチャンク化し続ける
├── flink/
│ └── astradb_writer.py # embeddings-sink を常時 consume して AstraDB に書き込み続ける
│ # ※ Flink SQL 自体は Confluent Cloud の SQL workspaces で実行
└── shared/
└── config.py # 全プロセス共通の環境変数設定
各プロセスは同一マシンでも別マシンでも動かせます。プロセス間の通信は Confluent(Kafka) が仲介するため、物理的な場所は問いません。
各プロセスの動作の違いは以下の通りです。
| プロセス | 動作 |
|---|---|
| Uploader | ブラウザで操作するたびに S3・Kafka に送信 |
| Extractor | 起動後はトピックを poll し続け、メッセージが来たら処理 |
| AstraDB Writer | 起動後はトピックを poll し続け、メッセージが来たら処理 |
| Flink SQL | Confluent Cloud 上で RUNNING 状態を維持し続ける |
各フェーズのコンポーネント解説（① Upload 〜 ⑤ ベクトル DB 書き込み）
1. S3 アップロード & Kafka への Claim 送信
uploader/app.py が担当します。
- INPUT：ユーザーがブラウザからアップロードした PDF ファイル
-
OUTPUT：
file-urisトピックに S3 URI イベントを publish
Streamlit の UI から PDF を受け取ったら、S3 へのアップロード → Kafka への通知 の2ステップで処理します。
ステップ①：クライアントの初期化
Confluent Cloud（Kafka）への接続には API キーと SASL/SSL 認証が必要です。confluent-kafka の Producer に設定辞書を渡して初期化します。S3 クライアントも同様に環境変数から認証情報を読み込みます。
from confluent_kafka import Producer
import boto3
# confluent-kafka の Producer を初期化
# Confluent Cloud は SASL_SSL + PLAIN 認証を使う
# bootstrap.servers はクラスターのエンドポイント（Confluent Cloud コンソールで確認できる）
producer = Producer({
"bootstrap.servers": os.environ["CONFLUENT_BOOTSTRAP_SERVERS"],
"security.protocol": "SASL_SSL", # 通信を TLS で暗号化
"sasl.mechanism": "PLAIN", # 認証方式
"sasl.username": os.environ["CONFLUENT_API_KEY"], # API キー
"sasl.password": os.environ["CONFLUENT_API_SECRET"], # API シークレット
})
# boto3 で S3 クライアントを初期化
s3 = boto3.client(
"s3",
aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
region_name=os.environ.get("AWS_REGION", "ap-northeast-1"),
)
BUCKET = os.environ["S3_BUCKET_NAME"]
ステップ②：S3 へアップロードして URI を取得
保存先のパスは uploads/日付/ファイルハッシュ/ファイル名 の形式にしています。ファイルハッシュをパスに含めることで、同じ内容のファイルが重複しても衝突せずに保存できます。
def upload_to_s3(file_bytes: bytes, filename: str, file_hash: str) -> str:
"""S3 に PDF をアップロードし、s3://バケット名/パス 形式の URI を返す。"""
key = f"uploads/{datetime.now(timezone.utc).strftime('%Y/%m/%d')}/{file_hash}/{filename}"
s3.put_object(Bucket=BUCKET, Key=key, Body=file_bytes, ContentType="application/pdf")
# 呼び出し元はこの URI だけ受け取り、Kafka に流す
return f"s3://{BUCKET}/{key}"
ステップ③：S3 URI だけを Kafka に publish（Claim Check の肝）
ファイル本体は S3 に置いたまま、Kafka には s3_uri だけを含む軽量な JSON を送ります。producer.produce() はバッファに積むだけなので、producer.flush() で実際に送信を確定させます。
def publish_event(s3_uri: str, filename: str, file_hash: str, file_size: int):
"""Kafka の file-uris トピックに URI イベントを publish する。"""
message = {
"message_id": str(uuid.uuid4()), # イベントの一意な識別子
"s3_uri": s3_uri, # ← ここに PDF の場所だけ入っている
"filename": filename,
"file_hash": file_hash,
"file_size_bytes": file_size,
"uploaded_at": datetime.now(timezone.utc).isoformat(),
}
producer.produce(
topic="file-uris",
key=file_hash, # 同じファイルのイベントは同じパーティションに送られる
value=json.dumps(message, ensure_ascii=False).encode("utf-8"),
)
producer.flush(timeout=10) # バッファをフラッシュして送信を確定
Kafka に publish される file-uris メッセージはこれだけです。PDF の実体は一切乗りません。
{
"message_id": "550e8400-e29b-41d4-a716-446655440000",
"s3_uri": "s3://rag-pipeline-bucket/uploads/2026/08/15/abc123/report.pdf",
"filename": "report.pdf",
"file_hash": "abc123...",
"file_size_bytes": 2048576,
"uploaded_at": "2026-08-15T09:00:00Z"
}
2. Extractor：S3 から取得して Docling で処理
extractor/app.py が担当します。
-
INPUT：
file-urisトピックの S3 URI イベント（セクション1の Uploader が publish） -
OUTPUT：
text-chunksトピックにテキストチャンクを publish
ステップ①：Consumer としてメッセージを待ち受ける
Extractor は file-uris トピックを購読する 常時起動プロセスです。Consumer の group.id は複数インスタンスを並列起動したときにメッセージの重複処理を防ぐための識別子です。auto.offset.reset: "earliest" は「まだ読んでいないメッセージがあれば先頭から読む」という設定で、プロセス再起動時に未処理分を取りこぼしません。
from confluent_kafka import Consumer, Producer
consumer = Consumer({
"bootstrap.servers": os.environ["CONFLUENT_BOOTSTRAP_SERVERS"],
"security.protocol": "SASL_SSL",
"sasl.mechanism": "PLAIN",
"sasl.username": os.environ["CONFLUENT_API_KEY"],
"sasl.password": os.environ["CONFLUENT_API_SECRET"],
"group.id": "text-extractor-group", # Consumer グループ名。同じ名前のプロセスは1件を1回だけ処理する
"auto.offset.reset": "earliest", # 未読メッセージを先頭から処理する
})
consumer.subscribe(["file-uris"]) # 購読するトピックを指定
# poll() でトピックを 1 秒間隔でポーリングし続ける
while True:
msg = consumer.poll(timeout=1.0) # メッセージがなければ 1 秒待って None を返す
if msg is None or msg.error():
continue # メッセージなし or エラーはスキップ
# メッセージ本文（bytes）を JSON にデコード
file_event = json.loads(msg.value().decode("utf-8"))
# ① S3 から PDF をダウンロード（Claim の引き換え）
pdf_bytes = download_pdf_from_s3(file_event["s3_uri"])
# ② Docling でテキスト抽出・チャンク化 → text-chunks に publish
chunks = extract_and_chunk(pdf_bytes, file_event["filename"])
publish_chunks(producer, file_event, chunks)
ステップ②：Docling で PDF を解析・チャンク化
Docling は IBM がオープンソースで開発している PDF 解析ライブラリです。DocumentConverter が PDF のレイアウト（見出し・本文・図・表）を解析し、HierarchicalChunker がセクション構造を保ったままチャンクに分割します。
単純な固定長分割（例：500文字ごとに切る）と違い、HierarchicalChunker は「章の途中で切れる」ような不自然な分割を避けます。また各チャンクには 見出しのパス（例: 1章 > 1.2節）と ページ番号 がメタデータとして付くため、RAG の回答に出典情報を添えることができます。
from docling.document_converter import DocumentConverter
from docling.chunking import HierarchicalChunker
# 起動時に1回だけ初期化（毎回生成するとモデルロードのオーバーヘッドがかかる）
_converter = DocumentConverter()
_chunker = HierarchicalChunker()
def extract_and_chunk(pdf_bytes: bytes, filename: str) -> list[dict]:
# Docling は URL か Path を受け取る API なので、bytes を一時ファイルに書き出す
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
tmp.write(pdf_bytes)
tmp_path = Path(tmp.name)
result = _converter.convert(tmp_path) # PDF を解析して DoclingDocument を生成
chunks = []
for chunk in _chunker.chunk(result.document): # セクション単位でチャンク化
text = chunk.text.strip()
if not text:
continue # 空チャンク（図のキャプションのみなど）はスキップ
# chunk.meta.headings は ["1章タイトル", "1.2節タイトル"] のような階層リスト
headings: list[str] = chunk.meta.headings or []
# このチャンクが含まれるページ番号を重複なく収集
page_numbers = sorted({
ref.page_no
for item in chunk.meta.doc_items
for ref in item.prov
if ref.page_no is not None
})
chunks.append({
"text": text,
"section_heading": " > ".join(headings) if headings else None, # 例: "1章 > 1.2節"
"page_numbers": page_numbers,
})
return chunks
チャンクは1件ずつ text-chunks トピックに publish します。各メッセージには元ファイルの file_hash・filename・s3_uri も含めており、後段で出典を追跡できるようにしています。
3. Flink SQL で Embedding をストリーミング生成
Confluent Cloud 上で動く Flink SQL ジョブが担当します。
-
INPUT：
text-chunksトピックのチャンク（セクション2の Extractor が publish） -
OUTPUT：
embeddings-sinkトピックにベクトル付きチャンクを publish
SQL workspaces で一度実行すると RUNNING 状態になり、text-chunks トピックにメッセージが届くたびに自動で Embedding を生成して embeddings-sink に書き出し続けます。
通常、テキストを Embedding（ベクトル）に変換するには Python コードで LLM API を呼び出し、結果を DB に書き込む処理を自前で実装する必要がありますが、Confluent Cloud Flink SQL の AI_EMBEDDING() 関数を使うことで、この処理を SQL だけで書けます。Kafka トピックに新しいメッセージが届くたびに Flink が自動で Embedding API を呼び出し、ベクトルを付与した結果を別のトピックに書き出してくれます。追加のアプリケーションコードは不要です。
ステップ①：Bedrock への接続とモデルを Flink に登録（初回のみ）
AI_EMBEDDING() を使うには、事前にどの AI サービス・モデルを使うかを Flink に登録しておく必要があります。Confluent Cloud のコンソールで左メニューの SQL workspaces を開き、ワークスペースを作成して以下の SQL を実行してください（初回のみ）。
-- Flink から AWS Bedrock に接続するための認証情報を登録する
CREATE CONNECTION bedrock_connection
TYPE AWS_BEDROCK
REGION 'us-east-1'
AWS_ACCESS_KEY_ID '...'
AWS_SECRET_ACCESS_KEY '...';
-- AI_EMBEDDING() で使うモデルを登録する
-- INPUT: チャンクのテキスト（STRING）を受け取り
-- OUTPUT: 1024 次元の浮動小数点数配列（= ベクトル）を返す
CREATE MODEL chunk_embedding
INPUT(text STRING)
OUTPUT(embedding ARRAY<FLOAT>)
WITH (
'TASK' = 'EMBEDDING',
'PROVIDER' = 'BEDROCK',
'BEDROCK.CONNECTION' = 'bedrock_connection',
'BEDROCK.MODEL_ID' = 'amazon.titan-embed-text-v2:0' -- Bedrock の Titan Embed v2 を使用
);
ステップ②：ストリーミング Embedding ジョブを起動
同じく SQL workspaces のエディタで以下の SQL を実行します。一度実行すると Flink ジョブが RUNNING 状態になり、text-chunks トピックに新しいメッセージが届くたびに自動で処理が走ります。
INSERT INTO `embeddings-sink`
SELECT
CAST(tc.chunk_id AS BYTES) AS `key`, -- Kafka メッセージのキー
tc.chunk_id,
CAST(tc.chunk_index AS BIGINT) AS chunk_index,
CAST(CURRENT_TIMESTAMP AS STRING) AS embedded_at, -- Embedding を付与した時刻
CAST(emb.embedding AS ARRAY<DOUBLE>) AS embedding, -- ← AI_EMBEDDING() が生成した 1024 次元ベクトル
tc.file_hash,
tc.filename,
CAST(tc.page_numbers AS ARRAY<BIGINT>) AS page_numbers,
tc.s3_uri,
tc.section_heading,
tc.`text`,
CAST(tc.total_chunks AS BIGINT) AS total_chunks
FROM `text-chunks` AS tc,
-- AI_EMBEDDING() はテーブル関数なので LATERAL TABLE で結合する（SELECT 内に直接書くとエラー）
LATERAL TABLE(AI_EMBEDDING('chunk_embedding', tc.`text`)) AS emb;
ハマりポイント：
AI_EMBEDDINGはスカラー関数（値を1つ返す関数）ではなくテーブル関数（行を返す関数）です。SELECT emb(...)のようにSELECT句に直接書くとエラーになります。必ずFROM句でLATERAL TABLE(AI_EMBEDDING(...)) AS embの形で結合してください。
4. AstraDB Writer：ベクトル DB への書き込み
flink/astradb_writer.py が担当します。
-
INPUT：
embeddings-sinkトピックのベクトル付きチャンク（セクション3の Flink SQL が publish） -
OUTPUT：AstraDB コレクションへの
insertOne
embeddings-sink トピックを常時 poll し続け、Flink が書き出したメッセージが届いたタイミングで AstraDB に書き込みます。
ステップ①：Confluent ワイヤフォーマットのデシリアライズ
Confluent Cloud Flink が Schema Registry 経由でトピックに書き込んだメッセージは、先頭 5 バイトにスキーマ ID が埋め込まれた Confluent ワイヤフォーマットになっています。そのままでは JSON として読めないため、JSONDeserializer を使って自動デコードします。
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.json_schema import JSONDeserializer
from confluent_kafka.serialization import SerializationContext, MessageField
# Schema Registry への接続（Confluent Cloud コンソールで確認できる）
sr_client = SchemaRegistryClient({
"url": os.environ["CONFLUENT_SR_URL"],
"basic.auth.user.info": f"{os.environ['CONFLUENT_SR_API_KEY']}:{os.environ['CONFLUENT_SR_API_SECRET']}",
})
deserializer = JSONDeserializer(None, schema_registry_client=sr_client)
ステップ②：embeddings-sink を consume して AstraDB に書き込む
AstraDB Writer も Extractor と同様に Consumer として常時起動します。embeddings-sink トピックからメッセージを受信したら、JSONDeserializer でデコードして AstraDB の Data API に insertOne で1件ずつ書き込みます。
consumer = Consumer({
**kafka_conf,
"group.id": "astradb-writer-group",
"auto.offset.reset": "latest", # Flink が生成したものだけ処理する。再起動時に全件再処理しない
})
consumer.subscribe(["embeddings-sink"])
while True:
msg = consumer.poll(timeout=1.0)
if msg is None or msg.error():
continue
# Confluent ワイヤフォーマット → dict へデシリアライズ
document = deserializer(
msg.value(),
SerializationContext("embeddings-sink", MessageField.VALUE),
)
insert_to_astradb(document)
ステップ③：AstraDB Data API に insertOne で書き込む
AstraDB への書き込みは REST API（Data API v2）を使います。ベクトルは $vector フィールドに渡すと、AstraDB がコレクション定義のインデックスと紐付けて保存します。
def insert_to_astradb(document: dict) -> None:
"""AstraDB Data API にチャンク1件を書き込む。"""
payload = {
"insertOne": {
"document": {
"_id": document["chunk_id"], # ドキュメントの一意な ID
"file_hash": document["file_hash"],
"filename": document["filename"],
"s3_uri": document["s3_uri"],
"chunk_index": document["chunk_index"],
"total_chunks": document["total_chunks"],
"text": document["text"], # 元テキスト（回答生成に使う）
"$vector": document["embedding"], # ← 1024 次元ベクトル。ここに入れると AstraDB がインデックス化する
"embedded_at": document["embedded_at"],
}
}
}
response = requests.post(
f"{ASTRADB_API_ENDPOINT}/{KEYSPACE}/{COLLECTION}",
headers={"Token": ASTRADB_TOKEN, "Content-Type": "application/json"},
json=payload,
timeout=15,
)
response.raise_for_status()
これで、PDF をアップロードしてから AstraDB にベクトルが保存されるまでのパイプラインが完成します。
Tips：Bob（AI コーディングアシスタント）× Confluent MCP サーバーで開発効率アップ
今回の開発で特に役立ったのが、IBM Bob（AI コーディングアシスタント）に Confluent の MCP サーバーを接続して自然言語で Confluent Cloud を操作するという手法です。
MCP（Model Context Protocol）は AI アシスタントが外部サービスのツールを直接呼び出せる仕組みです。Confluent が提供する MCP サーバーを Bob に接続することで、チャット画面から自然言語で Confluent Cloud を操作できます。
実際にこんな感じで使いました：
Confluent Cloud のインフラ操作
「text-chunks トピックのメッセージ件数を確認して」
→ Bob が get-partition-offsets ツールを呼び出して結果を返してくれる
「embedding-job-005 の Flink ステートメントが RUNNING かどうか確認して」
→ Bob が list-flink-statements を呼び出してステータスを教えてくれる
「embeddings-sink トピックの retention を 0 にしてパージして、その後 604800000ms に戻して」
→ Bob が alter-topic-config を2回呼び出してトピックをクリーンアップしてくれる
Flink SQL の生成
Flink SQL は独自の構文（LATERAL TABLE・CAST の書き方・型制約など）があり、ドキュメントを読みながら書くのは手間がかかります。Bob に「こういう処理をする Flink SQL を書いて」と伝えると、トピックのスキーマを MCP で確認しながら SQL を生成してくれます。
「text-chunks トピックのスキーマを確認して、embeddings-sink に AI_EMBEDDING で
ベクトルを付与して書き込む Flink SQL を書いて」
→ Bob が describe-flink-table で text-chunks のカラム一覧を取得し、
LATERAL TABLE(AI_EMBEDDING(...)) を使った INSERT INTO ... SELECT 文を生成してくれる
型のキャストや LATERAL TABLE の書き方など、ハマりやすい部分も Bob が自動で考慮してくれるため、SQL を一から書くより大幅に時間が短縮できました。
Confluent Cloud コンソールを開かずに、コーディング作業の合間にそのまま自然言語でインフラ操作・SQL 生成ができるのは非常に快適でした。特にデバッグ中の「トピックの中身を確認 → コード修正 → 再確認」というサイクルが格段に速くなります。
Bob への Confluent MCP サーバーの設定方法は以下の記事を参照してください。
実装する上での注意ポイント
| # | 問題 | 解決策 |
|---|---|---|
| 1 | Docling headings の型エラー |
最新版では list[str]。h.text ではなくそのまま join() する |
| 2 | Flink AI_EMBEDDING エラー |
テーブル関数なので LATERAL TABLE(...) で呼ぶ |
| 3 | Flink メッセージが読めない | Confluent ワイヤフォーマット → JSONDeserializer を使う |
| 4 | AstraDB $similarity エラー |
includeSimilarity: true は options に指定。projection には含めない |
| 5 | Flink の型不一致エラー |
CAST(emb.embedding AS ARRAY<DOUBLE>) など明示的にキャスト |
まとめ
Kafka で非構造化データを扱う際は、Claim Check パターンが実装の複雑さを抑えながら実用的なパイプラインを構築できる有効な選択肢です。
- Kafka には軽量な参照情報（URI）だけを流す
- ファイル本体は S3 などの外部ストレージで管理する
- Consumer が必要なときに S3 から取得して処理する
今回のパイプラインでは、PDF アップロードから AstraDB へのベクトル格納まで以下の4コンポーネントで実現しています。
| コンポーネント | 役割 |
|---|---|
| Uploader (Streamlit) | S3 へのアップロードと Kafka への URI 通知 |
| Extractor (Python) | S3 から PDF を取得し Docling でチャンク化 |
| Flink SQL |
AI_EMBEDDING() でストリーミング Embedding 生成 |
| AstraDB Writer (Python) | Embedding をベクトル DB に格納 |
特に強力だったのが Confluent Cloud Flink SQL の AI_EMBEDDING() 関数です。追加のアプリケーションコードなしに、ストリーミングデータに対してリアルタイムで Embedding API を適用できます。非構造化データを AI 活用パイプラインに乗せる際のシンプルで強力な選択肢です。
RAGなどにおいてリアルタイムにAIに与える情報を更新するニーズがある場合にはぜひこのようなパイプラインを実装してみてください。
参考リンク
- IBM Bob × Confluent MCP入門：KafkaとFlinkを自然言語で触ってみた
- Confluent Cloudで始めるApache Kafka & Flink入門
- Confluent Cloud API Key の Webコンソールでの作成方法
- Astra DB × Langflow × watsonx.ai：ノーコードでRAGアプリを構築してみた
- Confluent Cloud Flink AI_EMBEDDING ドキュメント
- Claim Check パターン（Enterprise Integration Patterns）
- Docling GitHub
- AstraDB Data API v2 ドキュメント
- AWS Bedrock Titan Embed v2