Cache hasil vektorisasi - Realtime Compute for Apache Flink

Dokumen ini menjelaskan cara mengonfigurasi tabel cache eksternal untuk fungsi AI_EMBED agar dapat menggunakan kembali hasil vektorisasi untuk teks input yang identik, sehingga mengurangi panggilan model, menurunkan latensi, dan menghemat biaya.

Informasi latar belakang

Fungsi AI_EMBED menggunakan parameter CACHE_TABLE untuk menentukan tabel eksternal sebagai cache persisten bagi hasil vektorisasi. Pada waktu proses, mekanisme ini bekerja sebagai berikut:

Cache hit: Jika pencarian asinkron menemukan kecocokan, fungsi langsung mengembalikan hasilnya tanpa memanggil model.
Cache miss: Jika pencarian asinkron tidak menemukan kecocokan, fungsi memanggil model untuk menghitung vektor, mengembalikan hasilnya, dan secara asinkron menuliskannya ke tabel cache.

Mekanisme ini mencegah panggilan model berulang untuk permintaan dengan teks input yang sama.

Batasan

Fitur ini hanya didukung oleh Realtime Compute for Apache Flink VVR 11.8.preview1 dan versi yang lebih baru.
Aktifkan Stream Storage Apache Fluss Edition. Tabel cache harus berupa tabel primary key Fluss.

Buat tabel cache

Tipe primary key dari tabel cache harus kompatibel dengan tipe CACHE_KEY.

Mode kunci yang ditentukan secara eksplisit

Tipe primary key harus kompatibel dengan tipe kolom yang ditentukan oleh CACHE_KEY. Misalnya, jika cache_key pada tabel sumber bertipe INT:

CREATE TABLE `fluss`.`default`.`ai_cache` (
  cache_key   INT NOT NULL,
  embedding   ARRAY<FLOAT>,PRIMARY KEY (cache_key) NOT ENFORCED
);

Mode kunci yang dihasilkan secara otomatis

Primary key harus bertipe STRING untuk menyimpan nilai hash SHA256:

CREATE TABLE `fluss`.`default`.`ai_cache_sha` (
  cache_key   STRING NOT NULL,
  embedding   ARRAY<FLOAT>,PRIMARY KEY (cache_key) NOT ENFORCED
);

Sintaks

AI_EMBED(
  MODEL       => MODEL <MODEL NAME>,
  INPUT       => <INPUT COLUMN NAME>
  [, DIMENSION   => <DIMENSION VALUE>]
  [, CACHE_TABLE => TABLE <catalog>.<db>.<table>
     [, CACHE_KEY => DESCRIPTOR(<column>)]]
  [, CONFIG      => MAP[...]]
)

Parameter

Parameter	Wajib	Deskripsi
`MODEL`	Ya	Referensi ke model. Untuk informasi selengkapnya, lihat Daftarkan model.
`INPUT`	Ya	Kolom teks input.
`CACHE_TABLE`	Tidak	Referensi ke tabel cache. Sintaksnya adalah `TABLE <catalog>.<db>.<table>`.
`CACHE_KEY`	Tidak	Kolom kunci cache. Sintaksnya adalah `DESCRIPTOR(<column>)`. Jika tidak ditentukan, sistem secara otomatis menghasilkan kunci komposit menggunakan hash SHA256.
`CONFIG`	Tidak	Konfigurasi waktu proses.

Mode CACHE_KEY

Yang ditentukan secara eksplisit (Direkomendasikan): Menggunakan nilai dari kolom tertentu pada tabel input sebagai kunci pencarian. Mode ini direkomendasikan jika data Anda sudah memiliki kunci unik.

CACHE_KEY => DESCRIPTOR(cache_key)

Pembuatan otomatis: Jika Anda tidak memberikan CACHE_KEY, sistem menghasilkan kunci berdasarkan aturan berikut:

-- Jika CACHE_KEY tidak diberikan, sistem secara otomatis menghasilkan kunci:
-- key = LOWER(SHA256(model_id || ':' || dimension || ':' || input_text))
CACHE_TABLE => TABLE `fluss-ql`.`default`.`ai_cache1`

Mode ini menimbulkan overhead perhitungan hash SHA256 pada waktu proses, tetapi secara otomatis memastikan kunci unik untuk setiap kombinasi model, dimensi, dan teks input.

Penyetelan performa

Parameter penyetelan

SET 'table.exec.async-ml-predict.max-concurrent-operations' = '100';
SET 'table.exec.async-ml-predict.output-mode' = 'ALLOW_UNORDERED';
SET 'table.exec.async-ml-predict.timeout' = '120s';

Parameter

Parameter	Bawaan	Deskripsi
`max-concurrent-operations`	10	Jumlah maksimum record untuk pencarian cache atau panggilan model secara konkuren. Meningkatkan nilai ini dapat meningkatkan throughput. Namun, pengaturan yang terlalu tinggi berpotensi membebani layanan model atau menyebabkan penggunaan memori berlebihan.
`output-mode`	`ORDERED`	Mode urutan output. `ORDERED`: Menjamin bahwa output memiliki urutan yang sama dengan input. Cocok untuk operasi downstream yang bergantung pada urutan, seperti join atau fungsi jendela. `ALLOW_UNORDERED`: Mengeluarkan record segera setelah selesai diproses. Direkomendasikan untuk aliran insert-only.
`timeout`	3min	Waktu total maksimum yang diizinkan untuk pencarian dan prediksi. Jika nilai ini terlalu rendah, operasi normal saat terjadi cache miss mungkin mengalami timeout.

Rekomendasi: Penggunaan ALLOW_UNORDERED memberikan manfaat performa signifikan karena alasan berikut:

Cache hit biasanya hanya memerlukan satu kali pencarian, sedangkan cache miss memerlukan panggilan model tambahan.
Pada mode ORDERED, satu cache miss menghambat output semua cache hit berikutnya.
Pada mode ALLOW_UNORDERED, cache hit langsung dikeluarkan tanpa menunggu penyelesaian cache miss sebelumnya.

Catatan

Hanya aliran insert-only yang menggunakan mode ALLOW_UNORDERED. Untuk aliran yang memiliki retraction, sistem akan kembali ke mode ORDERED, meskipun ALLOW_UNORDERED telah ditentukan.

Metrik pemantauan

Saat Pekerjaan sedang Berjalan, Anda dapat menggunakan metrik Flink untuk memantau efektivitas cache.

Nama metrik	Deskripsi	Catatan
`ai_function_cache.cache_hit`	Jumlah cache hit	Hit rate = hit / (hit + miss).
`ai_function_cache.cache_miss`	Jumlah cache miss	Nilai yang konsisten tinggi menunjukkan bahwa cache belum di-warm up atau kuncinya tidak stabil.
`ai_function_cache.writeback_attempt`	Jumlah upaya write-back	Nilai ini seharusnya mendekati jumlah cache miss.
`ai_function_cache.writeback_skipped_null_output`	Write-back yang dilewati karena output null	Nilai tinggi mengindikasikan potensi masalah pada model.

Contoh

Contoh tabel primary key Fluss:

-- Buat tabel cache (Fluss) terlebih dahulu
CREATE TABLE `fluss`.`default`.`embedding_cache` (
  cache_key   STRING NOT NULL,
  embedding   ARRAY<FLOAT>,
  PRIMARY KEY (cache_key) NOT ENFORCED
);

Contoh ini menggunakan tabel primary key Fluss sebagai cache dan mengatur mode output ke ALLOW_UNORDERED untuk meningkatkan throughput.

-- Parameter penyetelan
SET 'table.exec.async-ml-predict.max-concurrent-operations' = '100';
SET 'table.exec.async-ml-predict.output-mode' = 'ALLOW_UNORDERED';

-- Buat model
CREATE TEMPORARY MODEL embedding_model
  INPUT (text STRING)
  OUTPUT (embedding ARRAY<FLOAT>)
WITH (
  'provider' = 'openai-compat',
  'task' = 'embeddings',
  'model' = 'text-embedding-v4',
  'dimension' = '1024'
);

-- Kueri bisnis
INSERT INTO result_sink
SELECT id, content, embedding
FROM source_table, LATERAL TABLE(AI_EMBED(
  MODEL       => MODEL embedding_model,
  INPUT       => content,
  CONFIG      => MAP['async', 'true'],
  CACHE_TABLE => TABLE `fluss`.`default`.`embedding_cache`
));