Realtime Compute for Apache Flink:Referensi API Python DataFrame

Daftarkan fungsi skalar

Decorator udf mendaftarkan fungsi Python sebagai fungsi skalar untuk digunakan dalam DataFrame.

• Nama API: udf.

  udf(
      func=None,
      *,
      return_dtype=None,
      deterministic=True,
      name=None,
      func_type=None,
      concurrency=None,
      batch_size=None,
      num_gpus=None,
      gpu_type=None
  )

• Deskripsi: Mendaftarkan fungsi Python sebagai UDF skalar DataFrame. Mendukung fungsi biasa, fungsi async, subclass ScalarFunction / AsyncScalarFunction, dan kelas callable.

• Parameter

  Nama parameter	Tipe parameter	Wajib	Deskripsi
  func	Callable / Class	Tidak	Fungsi Python, instance/subclass ScalarFunction, atau kelas callable yang akan dibungkus. Jika dihilangkan, decorator dikembalikan.
  return_dtype	DataType / str / type	Tidak	Tipe kembalian. Dapat berupa instance DataType (misalnya, DataType.int64()), tipe Python (misalnya, int), atau string tipe SQL (misalnya, 'BIGINT'). Jika dihilangkan, akan diinferensi otomatis dari type hints fungsi.
  deterministic	Boolean	Tidak	Apakah fungsi tersebut deterministik. Default: True.
  name	String	Tidak	Nama UDF. Defaultnya adalah nama fungsi.
  func_type	String	Tidak	Format eksekusi. Opsi: "general", "pandas", "arrow". Jika dihilangkan, akan dideteksi otomatis dari type hints fungsi.
  concurrency	int	Tidak	Paralelisme operator UDF. UDF dengan nilai concurrency berbeda dipisahkan menjadi operator terpisah.
  batch_size	int	Tidak	Jumlah maksimum elemen per batch. Hanya berlaku untuk UDF batch (mode pandas / arrow).
  num_gpus	float	Tidak	Jumlah GPU yang diminta untuk UDF (misalnya, 0,5, 1). Saat diatur, UDF berjalan di operator terpisah dan tidak dirantai dengan UDF lain (termasuk UDF GPU lain).
  gpu_type	String	Tidak	Tipe GPU. Wajib ditentukan saat num_gpus disetel.

• Nilai kembalian

  Objek DataFrameUDFWrapper yang dapat digunakan dalam with_column, with_columns, map, map_batches, dan operasi serupa.

UDF baris sinkron

from pyflink.dataframe import col, udf

orders = pf.from_records(
    [
        (1001, 1, 99.9, "PAID"),
        (1002, 2, 35.5, "CREATED"),
        (1003, 1, 188.0, "PAID"),
        (1004, 3, 88.8, None),
    ],
    schema=["order_id", "user_id", "amount", "status"],
)

@udf
def normalize_status(status: str) -> str:
    if status is None:
        return "UNKNOWN"
    return status.strip().upper()

@udf(concurrency=32)
def order_tag(amount: float, status: str) -> str:
    """Mengambil beberapa kolom sebagai input"""
    if status == "PAID" and amount is not None and amount >= 100:
        return "high_value_paid"
    return "normal"

@udf
class ModelPredictor:
    def __init__(self):
        # Inisialisasi dieksekusi di task manager
        self._model = load_model()

    def __call__(self, amount: float) -> str:
        return self._model.predict(amount)

# with_column menambahkan satu kolom
orders_with_status = orders.with_column(
    "status_norm", normalize_status(col("status"))
)

# with_columns menambahkan beberapa kolom sekaligus
orders_with_flags = orders.with_columns(
    status_norm=normalize_status(col("status")),
    tag=order_tag(col("amount"), col("status")),
)

# Gunakan UDF stateful
orders_with_predictions = orders.with_column(
    "prediction", ModelPredictor(col("amount"))
)

def build_order_feature(row):
    amount = row["amount"] or 0.0
    return {
        "order_id": row["order_id"],
        "user_id": row["user_id"],
        "feature": f"{row['status']}:{'large' if amount >= 100 else 'normal'}",
    }

order_features = orders.map(
    build_order_feature,
    return_dtype=DataType.struct({
        "order_id": DataType.int64(),
        "user_id": DataType.int64(),
        "feature": DataType.string(),
    }),
)

from typing import TypedDict

class OrderFeature(TypedDict):
    order_id: int
    user_id: int
    feature: str

def build_order_feature_typed(row) -> OrderFeature:
    amount = row["amount"] or 0.0
    return {
        "order_id": row["order_id"],
        "user_id": row["user_id"],
        "feature": f"{row['status']}:{'large' if amount >= 100 else 'normal'}",
    }

order_features_typed = orders.map(build_order_feature_typed)

UDF vektor sinkron

import pandas as pd

@udf(return_dtype=DataType.float64(), concurrency=32, batch_size=64)
def scale_amount_pandas(amounts: pd.Series) -> pd.Series:
    return amounts * 100.0

orders_scaled_pandas = orders.with_column(
    "amount_scaled", scale_amount_pandas(col("amount")),
)

import pyarrow as pa
import pyarrow.compute as pc

@udf(return_dtype=DataType.float64(), concurrency=32, batch_size=64)
def scale_amount_arrow(amounts: pa.Array) -> pa.Array:
    return pc.multiply(amounts, 100.0)

orders_scaled_arrow = orders.with_column(
    "amount_scaled", scale_amount_arrow(col("amount")),
)

def score_batch_pandas(batch: dict[str, pd.Series]) -> dict[str, pd.Series]:
    amount = batch["amount"].fillna(0.0)
    return {
        "order_id": batch["order_id"],
        "score": (amount / 100.0).clip(0.0, 1.0),
    }

order_scores_pandas = orders.map_batches(
    score_batch_pandas,
    batch_format="pandas",
    batch_size=1024,
    return_dtype=DataType.struct(
        {
            "order_id": DataType.int64(),
            "score": DataType.float64(),
        }
    ),
)

def score_batch_arrow(batch: dict[str, pa.Array]) -> dict[str, pa.Array]:
    amount = pc.if_else(pc.is_null(batch["amount"]), 0.0, batch["amount"])
    raw_score = pc.divide(pc.cast(amount, pa.float64()), 100.0)
    score = pc.if_else(
        pc.less(raw_score, 0.0),
        0.0,
        pc.if_else(pc.greater(raw_score, 1.0), 1.0, raw_score),
    )
    return {
        "order_id": batch["order_id"],
        "score": score,
    }

order_scores_arrow = orders.map_batches(
    score_batch_arrow,
    batch_format="arrow",
    batch_size=1024,
    return_dtype=DataType.struct(
        {
            "order_id": DataType.int64(),
            "score": DataType.float64(),
        }
    ),
)

UDF baris asinkron

UDF asinkron memungkinkan beberapa baris melakukan operasi I/O secara bersamaan, yang meningkatkan throughput. Fungsi ini cocok untuk logika yang memanggil layanan eksternal atau yang intensif I/O. Anda dapat memanggil UDF asinkron berbasis baris di dalam <a href="https://ververica-flink-docs.readthedocs.io/en/latest/reference/pyflink.dataframe/api/pyflink.dataframe.DataFrame.with_column.html#pyflink.dataframe.DataFrame.with_column" id="a4862ccb65o83">with_column</a>, <a href="https://ververica-flink-docs.readthedocs.io/en/latest/reference/pyflink.dataframe/api/pyflink.dataframe.DataFrame.with_columns.html#pyflink.dataframe.DataFrame.with_columns" id="1b48944541kmy">with_columns</a>, atau map.

import asyncio

# Gunakan di `with_column`
@udf(concurrency=32)
async def query_region(user_id: int) -> str:
    await asyncio.sleep(0.01)  # simulasi I/O asinkron
    return f"region_for_{user_id}"

orders_with_region = orders.with_column(
    "region", query_region(col("user_id")),
)

# Gunakan di `map`
class OrderWithRegion(TypedDict):
    order_id: int
    region: str

async def enrich_order_region(row) -> OrderWithRegion:
    await asyncio.sleep(0.01)  # Simulasi I/O asinkron
    return {
        "order_id": row["order_id"],
        "region": f"region_for_{row['user_id']}",
    }

orders_with_region = orders.map(enrich_order_region)

UDF vektor asinkron

UDF vektorisasi asinkron menggabungkan throughput pemrosesan batch dengan konkurensi I/O asinkron. Gunakan batch_size untuk mengontrol ukuran setiap batch. Anda dapat memanggil UDF vektorisasi asinkron di with_column, with_columns, atau map_batches.

# Gunakan di `with_column`
@udf(return_dtype=DataType.string(), concurrency=32, batch_size=64)
async def batch_enrich(statuses: pd.Series) -> pd.Series:
    async def enrich_one(s):
        await asyncio.sleep(0.01)  # simulasi panggilan API asinkron
        return f"enriched_{s}"
    tasks = [enrich_one(s) for s in statuses]
    results = await asyncio.gather(*tasks)
    return pd.Series(results)

orders_enriched = orders.with_column(
    "status_enriched", batch_enrich(col("status")),
)

# Gunakan di `map_batches`
async def enrich_order_batch(
    batch: dict[str, pd.Series],
) -> dict[str, pd.Series]:
    async def query_region(user_id):
        await asyncio.sleep(0.01)  # Simulasikan I/O asinkron
        return f"region_for_{user_id}"

    regions = await asyncio.gather(
        *(query_region(user_id) for user_id in batch["user_id"])
    )

    return {
        "order_id": batch["order_id"],
        "region": pd.Series(
            regions,
            index=batch["user_id"].index,
        ),
    }

orders_with_region = orders.map_batches(
    enrich_order_batch,
    batch_format="pandas",
    batch_size=64,
    concurrency=32,
    return_dtype=DataType.struct({
        "order_id": DataType.int64(),
        "region": DataType.string(),
    }),
)

Daftarkan fungsi tabel

Decorator udtf mendaftarkan fungsi Python sebagai fungsi tabel untuk digunakan dalam DataFrame.

• Nama API: udtf.

  udtf(
      func=None,
      *,
      return_dtype=None,
      deterministic=True,
      name=None,
      concurrency=None,
      num_gpus=None,
      gpu_type=None
  )

• Deskripsi: Mendaftarkan fungsi Python, subclass atau instance TableFunction, atau kelas callable sebagai fungsi tabel untuk digunakan dalam DataFrame.

• Parameter

  Nama parameter	Tipe parameter	Wajib	Deskripsi
  func	Callable / Class	Tidak	Fungsi Python, instance/subclass TableFunction, atau kelas callable yang akan dibungkus. Jika dihilangkan, decorator dikembalikan.
  return_dtype	DataType / str / type	Tidak	Tipe setiap baris output. Untuk output multi-kolom, gunakan DataType.struct({...}) atau TypedDict.
  deterministic	Boolean	Tidak	Apakah fungsi tersebut deterministik. Default: True.
  name	String	Tidak	Nama UDTF. Defaultnya adalah nama fungsi.
  concurrency	int	Tidak	Paralelisme operator UDTF.
  num_gpus	float	Tidak	Jumlah GPU yang diminta untuk UDTF (misalnya, 0,5, 1). Saat diatur, UDTF berjalan di operator terpisah dan tidak dirantai dengan UDF lain (termasuk UDF GPU lain).
  gpu_type	String	Tidak	Tipe GPU. Wajib ditentukan saat num_gpus disetel.

• Nilai kembalian

  Objek DataFrameUDTFWrapper yang dapat digunakan dalam flat_map atau join_lateral.

Di join_lateral Digunakan di

join_lateral menambahkan hasil UDTF setelah baris input asli. Secara default, baris input yang tidak menghasilkan output UDTF dihapus; Anda dapat mengatur ignore_empty=False untuk menyimpan baris input tanpa output UDTF.

from typing import Iterator, Tuple
from pyflink.dataframe import udtf

# Gunakan type hint dan .alias() untuk menentukan tipe kembalian dan nama kolom
@udtf
def split_words(text: str) -> Iterator[Tuple[str, int]]:
    for word in text.split():
        yield word, len(word)

words_with_source = texts.join_lateral(
    split_words(col("text")).alias("word", "word_length"),
    ignore_empty=False,
)

# Atau gunakan return_dtype
@udtf(return_dtype=DataType.struct({
    "word": DataType.string(),
    "word_length": DataType.int32(),
}))
def split_words_with_length(text):
    for word in text.split():
        yield {"word": word, "word_length": len(word)}

words_with_detail = texts.join_lateral(
    split_words_with_length(col("text")),
)

Di flat_map Digunakan di

<a href="https://ververica-flink-docs.readthedocs.io/en/latest/reference/pyflink.dataframe/api/pyflink.dataframe.DataFrame.flat_map.html#pyflink.dataframe.DataFrame.flat_map" id="35bfeae711bka">flat_map</a> memanggil UDTF pada setiap baris. Hasilnya hanya berisi kolom output UDTF, dan kolom input asli tidak dipertahankan. Anda dapat meneruskan fungsi Python secara langsung ke flat_map tanpa membungkusnya dengan decorator udtf.

from typing import Any, Dict, Iterator, TypedDict

# Gunakan type hint dengan TypedDict untuk menentukan tipe kembalian dan nama kolom
class Word(TypedDict):
    word: str
    length: int

def split_row(row: Dict[str, Any]) -> Iterator[Word]:
    for word in row["text"].split():
        yield {"word": word, "length": len(word)}

words = texts.flat_map(split_row)

# Atau gunakan return_dtype
def split_row_with_dtype(row):
    for word in row["text"].split():
        yield {"word": word}

words_with_dtype = texts.flat_map(
    split_row_with_dtype,
    return_dtype=DataType.struct({"word": DataType.string()}),
)

Klasifikasi teks

• Nama API: ai_classify.

  DataFrame.llm.ai_classify(
      input_col,
      labels,
      *,
      provider=None,
      model=None,
      cache_table=None,
      cache_key=None,
      config=None,
      **kwargs
  )

• Deskripsi: Mengklasifikasikan teks ke salah satu label yang ditentukan.

• Parameter

  Nama parameter	Tipe parameter	Wajib	Deskripsi
  input_col	String / Expression	Ya	Nama kolom teks input atau ekspresi kolom.
  labels	List[String]	Ya	Daftar label klasifikasi, misalnya, ["positive", "negative", "neutral"].

  Lihat deskripsi parameter lain di Pemanggilan generik.

  Catatan

  Fungsi AI spesifik domain seperti ai_classify menggunakan system prompt bawaan. System prompt yang dikonfigurasi pada Model Provider tidak berlaku.

• Nilai kembalian

  DataFrame dengan kolom category (STRING) dan confidence (DOUBLE) ditambahkan, yang berisi hasil klasifikasi dan skor kepercayaan.

• Contoh

  pf.set_model_provider(pf.OpenAICompatProvider(task="chat/completions"))
  df = pf.from_dict({"review": ["Great product", "Terrible, do not buy", "It is okay"]})
  df = df.llm.ai_classify("review",
                          labels=["positive", "negative", "neutral"],
                          model="qwen3.6-plus")

Analisis sentimen

• Nama API: ai_sentiment.

  DataFrame.llm.ai_sentiment(
      input_col,
      *,
      provider=None,
      model=None,
      cache_table=None,
      cache_key=None,
      config=None,
      **kwargs
  )

• Deskripsi: Melakukan analisis sentimen pada teks input.

• Parameter

  Nama parameter	Tipe parameter	Wajib	Deskripsi
  input_col	String / Expression	Ya	Nama kolom teks input atau ekspresi kolom.

  Lihat deskripsi parameter lain di Pemanggilan generik.

• Nilai kembalian

  DataFrame dengan kolom berikut ditambahkan, yang berisi hasil analisis sentimen:

  • score (DOUBLE): Skor analisis sentimen, berkisar dari -1,0 hingga 1,0.

  • label (STRING): Salah satu dari "positive", "negative", atau "neutral".

  • confidence (DOUBLE): Skor kepercayaan.

• Contoh

  pf.set_model_provider(pf.OpenAICompatProvider(task="chat/completions"))
  df = pf.from_dict({"comment": ["This feature is amazing!", "Broke after one month"]})
  df = df.llm.ai_sentiment("comment", model="qwen3.6-plus")

Ekstraksi informasi

• Nama API: ai_extract.

  DataFrame.llm.ai_extract(
      input_col,
      schema,
      *,
      provider=None,
      model=None,
      cache_table=None,
      cache_key=None,
      config=None,
      **kwargs
  )
  

• Deskripsi: Mengekstraksi informasi terstruktur dari teks berdasarkan skema JSON yang diberikan.

• Parameter

  Nama parameter	Tipe parameter	Wajib	Deskripsi
  input_col	String / Expression	Ya	Nama kolom teks input atau ekspresi kolom.
  schema	String	Ya	String skema JSON yang mendeskripsikan bidang yang akan diekstraksi, misalnya, '{"name":"STRING", "phone":"STRING"}'.

  Lihat deskripsi parameter lain di Pemanggilan generik.

• Nilai kembalian

  DataFrame dengan kolom extracted_json (STRING) ditambahkan, yang berisi informasi terstruktur yang diekstraksi.

• Contoh

  pf.set_model_provider(pf.OpenAICompatProvider(task="chat/completions"))
  df = pf.from_dict({"text": ["John Smith, male, 28 years old, phone ***-****"]})
  schema = '{"name": "STRING", "age": "INTEGER", "phone": "STRING"}'
  df = df.llm.ai_extract("text", schema=schema,
                         model="qwen3.6-plus")

Terjemahan teks

• Nama API: ai_translate.

  DataFrame.llm.ai_translate(
      input_col,
      source_lang,
      target_lang,
      *,
      provider=None,
      model=None,
      cache_table=None,
      cache_key=None,
      config=None,
      **kwargs
  )

• Deskripsi: Menerjemahkan teks dari bahasa sumber ke bahasa target.

• Parameter

  Nama parameter	Tipe parameter	Wajib	Deskripsi
  input_col	String / Expression	Ya	Nama kolom teks input atau ekspresi kolom.
  source_lang	String	Ya	Kode bahasa sumber, misalnya: "zh", "en", "auto" (terdeteksi otomatis). Didukung: auto, zh, en, ja, ko, fr, de, es, ru, ar, pt.
  target_lang	String	Ya	Kode bahasa target. Nilai ini tidak boleh "auto".

  Lihat deskripsi parameter lain di Pemanggilan generik.

• Nilai kembalian

  DataFrame dengan kolom translated_text (STRING) dan detected_language (STRING) ditambahkan, yang berisi teks terjemahan dan bahasa yang terdeteksi.

• Contoh

  pf.set_model_provider(pf.OpenAICompatProvider(task="chat/completions"))
  df = pf.from_dict({"text": ["Hello World", "How are you?"]})
  df = df.llm.ai_translate("text", source_lang="en", target_lang="zh",
                           model="qwen3.6-plus")

Ringkasan teks

• Nama API: ai_summarize.

  DataFrame.llm.ai_summarize(
      input_col,
      max_length,
      *,
      provider=None,
      model=None,
      cache_table=None,
      cache_key=None,
      config=None,
      **kwargs
  )
  

• Deskripsi: Merangkum teks hingga panjang maksimum yang ditentukan.

• Parameter

  Nama parameter	Tipe parameter	Wajib	Deskripsi
  input_col	String / Expression	Ya	Nama kolom teks input atau ekspresi kolom.
  max_length	int	Ya	Jumlah maksimum karakter dalam ringkasan. Harus lebih besar dari 0.

  Lihat deskripsi parameter lain di Pemanggilan generik.

• Nilai kembalian

  DataFrame dengan kolom summary (STRING) ditambahkan, yang berisi teks ringkasan.

• Contoh

  pf.set_model_provider(pf.OpenAICompatProvider(task="chat/completions"))
  df = pf.from_dict({"article": ["This is a long article with lots of content..."]})
  df = df.llm.ai_summarize("article", max_length=100,
                           model="qwen3.6-plus")
  

Penyembunyian data

• Nama API: ai_mask.

  DataFrame.llm.ai_mask(
      input_col,
      entities,
      *,
      provider=None,
      model=None,
      cache_table=None,
      cache_key=None,
      config=None,
      **kwargs
  )
  

• Deskripsi: Menyembunyikan informasi sensitif dalam teks.

• Parameter

  Nama parameter	Tipe parameter	Wajib	Deskripsi
  input_col	String / Expression	Ya	Nama kolom teks input atau ekspresi kolom.
  entities	List[String]	Ya	Daftar tipe entitas yang akan disembunyikan, misalnya, ["name", "phone"].

  Lihat deskripsi parameter lain di Pemanggilan generik.

• Nilai kembalian

  DataFrame dengan kolom masked_text (STRING) dan detected_entities (ARRAY<STRING>) ditambahkan, yang berisi teks yang telah disembunyikan dan entitas yang terdeteksi.

• Contoh

  pf.set_model_provider(pf.OpenAICompatProvider(task="chat/completions"))
  df = pf.from_dict({"text": ["Please contact John Smith at 555-0123"]})
  df = df.llm.ai_mask("text", entities=["name", "phone"],
                      model="qwen3.6-plus")

Pencarian vektor

• Nama API: vector_search.

  DataFrame.llm.vector_search(
      search_source,
      column_to_search,
      column_to_query,
      top_k,
      *,
      agg=False,
      output_columns=None,
      config=None,
      ignore_empty=False
  )

• Deskripsi: Mencari sumber pencarian vektor untuk catatan paling mirip menggunakan vektor kueri di DataFrame saat ini, dan menambahkan hasil pencarian ke DataFrame saat ini.

• Parameter

  Nama parameter	Tipe parameter	Wajib	Deskripsi
  search_source	DataFrame	Ya	DataFrame sumber pencarian vektor, misalnya, koleksi vektor yang dibaca menggunakan read_milvus.
  column_to_search	String	Ya	Nama kolom vektor di sumber pencarian.
  column_to_query	String / Expression	Ya	Nama kolom vektor kueri atau ekspresi kolom di DataFrame saat ini.
  top_k	int	Ya	Jumlah catatan paling mirip yang dikembalikan per catatan input.
  agg	Boolean	Tidak	Menentukan apakah hasil top K diagregasi ke dalam satu kolom array. Nilai default: False.
  output_columns	List[String] / Mapping[String, String]	Tidak	Kolom hasil pencarian vektor yang akan dikembalikan. Daftar memilih kolom dan mempertahankan nama aslinya. Dengan agg=False Anda juga dapat menggunakan pemetaan, di mana kunci memilih kolom dan nilai menjadi alias output. Anda dapat memilih kolom tingkat atas search_source dan kolom score yang dihasilkan oleh engine. Nilai default: None, yang memilih semua kolom hasil.
  config	Dict	Tidak	Konfigurasi pencarian vektor waktu proses.
  ignore_empty	Boolean	Tidak	Menentukan apakah baris input yang tidak memiliki hasil pencarian dihapus. Nilai default: False.

• Nilai kembalian

  Objek DataFrame yang berisi kolom asli dan kolom hasil pencarian vektor yang ditambahkan. Asumsikan bahwa output_columns memilih total M kolom hasil:

  • Jika Anda tidak menentukan output_columns, semua kolom search_source dan kolom skor pencarian bertipe DOUBLE dikembalikan. Secara default kolom skor dinamai score. Makna dan arah pengurutan skor bergantung pada connector dan metrik jaraknya. Untuk Milvus, interpretasikan nilai tersebut bersama dengan search_metric.

  • Saat agg=False, setiap baris input menghasilkan paling banyak top_k baris output dengan M kolom ditambahkan. Daftar memilih kolom dalam urutan yang dideklarasikan dan mempertahankan nama aslinya. Pemetaan memilih kolom dalam urutan deklarasi kuncinya dan menggunakan nilai yang sesuai sebagai alias output.

  • Saat `agg=True`, setiap input yang memiliki hasil pencarian menghasilkan paling banyak satu baris output, dengan kolom search_results bertipe ARRAY<ROW<...>> ditambahkan. Secara default ignore_empty=False menyimpan semua catatan input, dan search_results bernilai NULL saat input tidak memiliki hasil pencarian. Jika Anda mengatur ignore_empty=True, input tanpa hasil pencarian tidak muncul dalam hasil. Setiap elemen array adalah ROW yang berisi M bidang yang dipilih. Mode ini hanya mendukung daftar untuk memproyeksikan bidang dalam, bukan pemetaan. Setelah pemanggilan, ganti nama kolom array luar dengan memanggil `rename_columns("search_results", "new_name")`.

  • output_columns tidak boleh kosong, dan baik nama kolom maupun alias tidak boleh kosong atau duplikat. Kolom yang dipilih harus merupakan kolom tingkat atas search_source atau kolom kemiripan yang dihasilkan. Urutan yang dikembalikan sesuai dengan urutan deklarasi daftar atau kunci pemetaan.

  • Jika search_source sudah memiliki kolom seperti score atau score0, engine memilih nama unik untuk kolom kemiripan yang dihasilkan, seperti score2. Dalam kasus ini, gunakan nama kolom aktual di output_columns.

  • Dalam mode non-agregasi, Anda dapat menggunakan pemetaan untuk memberikan alias kolom hasil yang tidak bentrok. Jika daftar atau output default bentrok dengan kolom DataFrame saat ini, ganti nama kolom yang bentrok dengan rename_columns sebelum memanggil vector_search. Dalam mode agregasi, jika DataFrame saat ini sudah memiliki kolom search_results, ganti nama kolom tersebut terlebih dahulu.

  • Jika connector sumber pencarian vektor mendukung pushdown pemangkasan kolom, engine mendorong pemangkasan output_columns ke connector. Terlepas dari apakah pushdown didukung atau tidak, hasil akhir hanya berisi kolom hasil pencarian vektor yang dipilih.

• Catatan tentang peningkatan dari VVR 11.8

  Di VVR 11.8, output_columns berarti alias semua kolom output. Di VVR 11.9 dan versi setelahnya, artinya kolom hasil yang dipilih. Misalnya, pemanggilan berikut yang ditulis untuk VVR 11.8:

  output_columns=["match_doc_id", "match_embedding", "match_title", "match_score"]

  Setelah peningkatan, gunakan pemetaan sebagai gantinya:

  output_columns={
      "doc_id": "match_doc_id",
      "embedding": "match_embedding",
      "title": "match_title",
      "score": "match_score",
  }

• Contoh

  from array import array
  import pyflink.dataframe as pf
  from pyflink.dataframe import DataType
  
  query_df = pf.from_dict({
      "query_id": [1],
      "query_embedding": [array("f", [0.1, 0.2, 0.3])],
  })
  
  documents = pf.read_milvus(
      endpoint="http://milvus.example.com",
      username="${secret_values.milvus_user}",
      password="${secret_values.milvus_password}",
      database_name="commerce",
      collection_name="support_docs",
      schema={
          "doc_id": DataType.int64(),
          "embedding": DataType.list(DataType.float32()),
          "title": DataType.string(),
      },
      search_metric="COSINE",
  )
  
  selected_docs = query_df.llm.vector_search(
      documents,
      column_to_search="embedding",
      column_to_query="query_embedding",
      top_k=3,
      output_columns=["doc_id", "title", "score"],
  )

  Daftar memproyeksikan kolom hasil dan mempertahankan nama aslinya. Untuk menentukan alias sekaligus, gunakan pemetaan:

  matched_docs = query_df.llm.vector_search(
      documents,
      column_to_search="embedding",
      column_to_query="query_embedding",
      top_k=3,
      output_columns={
          "doc_id": "match_doc_id",
          "title": "match_title",
          "score": "match_score",
      },
  )

  Dalam mode agregasi, daftar memilih bidang setiap elemen array, dan kolom array luar selalu dinamai search_results. Untuk mengganti nama kolom array luar, panggil rename_columns:

  matched_docs_agg = query_df.llm.vector_search(
      documents,
      column_to_search="embedding",
      column_to_query="query_embedding",
      top_k=3,
      agg=True,
      output_columns=["doc_id", "title", "score"],
  ).rename_columns("search_results", "matches")

Embedding teks

• Nama API: ai_embed.

  DataFrame.llm.ai_embed(
      input_col,
      dimension=1024,
      *,
      provider=None,
      model=None,
      cache_table=None,
      cache_key=None,
      config=None,
      **kwargs
  )
  

• Deskripsi: Menghasilkan embedding vektor untuk teks.

• Parameter

  Nama parameter	Tipe parameter	Wajib	Deskripsi
  input_col	String / Expression	Ya	Nama kolom teks input atau ekspresi kolom.
  dimension	int	Tidak	Dimensi vektor. Nilai default: 1024.

  Lihat deskripsi parameter lain di Pemanggilan generik.

• Nilai kembalian

  DataFrame dengan kolom embedding (ARRAY<FLOAT>) ditambahkan, yang berisi vektor yang dihasilkan.

• Contoh

  pf.set_model_provider("embedding", pf.OpenAICompatProvider(task="embeddings"))
  df = pf.from_dict({"text": ["stream processing with Flink", "real-time analytics"]})
  df = df.llm.ai_embed("text", dimension=1024,
                       provider="embedding", model="text-embedding-v4")