Realtime Compute for Apache Flink:Python DataFrame API 参考

Skaler fonksiyonları kaydetme

udf dekoratörü, bir Python fonksiyonunu DataFrames içinde kullanılmak üzere skaler bir fonksiyon olarak kaydeder.

• API adı: 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
  )

• Açıklama: Bir Python fonksiyonunu DataFrame skaler UDF olarak kaydeder. Normal fonksiyonları, asenkron fonksiyonları, ScalarFunction / AsyncScalarFunction alt sınıflarını ve çağrılabilir sınıfları destekler.

• Parametreler

  Parametre adı	Parametre türü	Zorunlu	Açıklama
  func	Callable / Sınıf	Hayır	Sarmalanacak Python fonksiyonu, ScalarFunction örneği/alt sınıfı veya çağrılabilir sınıf. Atlanırsa bir dekoratör döndürülür.
  return_dtype	DataType / str / type	Hayır	Dönüş türü. DataType örneği (örneğin, DataType.int64()), Python türleri (örneğin, int) veya SQL türü dizeleri (örneğin, 'BIGINT') olabilir. Atlanırsa, fonksiyon türü ipuçlarından otomatik olarak çıkarılır.
  deterministic	Boolean	Hayır	Fonksiyonun deterministik olup olmadığı. Varsayılan: True.
  name	Dize	Hayır	UDF adı. Varsayılan olarak fonksiyon adını alır.
  func_type	Dize	Hayır	Çalıştırma formatı. Seçenekler: "general", "pandas", "arrow". Atlanırsa, fonksiyon türü ipuçlarından otomatik olarak algılanır.
  concurrency	int	Hayır	UDF operatörünün paralellik değeri. Farklı eşzamanlılık değerlerine sahip UDF'ler ayrı operatörlere ayrılır.
  batch_size	int	Hayır	Toplu iş başına maksimum öğe sayısı. Yalnızca toplu iş UDF'leri (pandas / arrow modu) için geçerlidir.
  num_gpus	float	Hayır	UDF için istenecek GPU sayısı (örneğin, 0,5, 1). Ayarlandığında, UDF ayrı bir operatörde çalışır ve diğer UDF'lerle (diğer GPU UDF'leri dahil) zincirlenmez.
  gpu_type	Dize	Hayır	GPU türü. num_gpus belirtildiğinde gereklidir.

• Dönüş değeri

  with_column, with_columns, map, map_batches ve benzeri işlemlerde kullanılabilecek bir DataFrameUDFWrapper nesnesi.

Senkron satır UDF

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:
    """Takes multiple columns as input"""
    if status == "PAID" and amount is not None and amount >= 100:
        return "high_value_paid"
    return "normal"

@udf
class ModelPredictor:
    def __init__(self):
        # Initialization is executed on the task manager
        self._model = load_model()

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

# with_column adds a single column
orders_with_status = orders.with_column(
    "status_norm", normalize_status(col("status"))
)

# with_columns adds multiple columns at once
orders_with_flags = orders.with_columns(
    status_norm=normalize_status(col("status")),
    tag=order_tag(col("amount"), col("status")),
)

# Use stateful UDFs
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)

Senkron vektör UDF

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(),
        }
    ),
)

Asenkron satır UDF

Asenkron UDF'ler, birden çok satırın G/Ç işlemlerini eş zamanlı olarak gerçekleştirmesine olanak tanıyarak aktarım hızını artırır. Harici hizmetleri çağıran veya G/Ç yoğunluklu mantık içeren işlemler için uygundur. Asenkron satır tabanlı UDF'leri with_column, with_columns veya map içinde çağırabilirsiniz.

import asyncio

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

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

# Use in `map`
class OrderWithRegion(TypedDict):
    order_id: int
    region: str

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

orders_with_region = orders.map(enrich_order_region)

Asenkron vektör UDF

Asenkron vektörize UDF'ler, toplu işleme aktarım hızını asenkron G/Ç eşzamanlılığı ile birleştirir. Her bir toplu işin boyutunu kontrol etmek için batch_size parametresini kullanın. Asenkron vektörize UDF'leri with_column, with_columns veya map_batches içinde çağırabilirsiniz.

# Use in `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)  # simulate async API call
        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")),
)

# Use in `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)  # Simulate asynchronous I/O
        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(),
    }),
)

Tablo fonksiyonlarını kaydetme

udtf dekoratörü, bir Python fonksiyonunu DataFrames içinde kullanılmak üzere tablo fonksiyonu olarak kaydeder.

• API adı: udtf.

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

• Açıklama: Bir Python fonksiyonunu, TableFunction alt sınıfını/örneğini veya çağrılabilir bir sınıfı, DataFrames içinde kullanılmak üzere tablo fonksiyonu olarak kaydeder.

• Parametreler

  Parametre adı	Parametre türü	Zorunlu	Açıklama
  func	Callable / Sınıf	Hayır	Sarmalanacak Python fonksiyonu, TableFunction örneği/alt sınıfı veya çağrılabilir sınıf. Atlanırsa bir dekoratör döndürülür.
  return_dtype	DataType / str / type	Hayır	Her çıktı satırının türü. Çok sütunlu çıktılar için DataType.struct({...}) veya TypedDict kullanın.
  deterministic	Boolean	Hayır	Fonksiyonun deterministik olup olmadığı. Varsayılan: True.
  name	Dize	Hayır	UDTF adı. Varsayılan olarak fonksiyon adını alır.
  concurrency	int	Hayır	UDTF operatörünün paralellik değeri.
  num_gpus	float	Hayır	UDTF için istenecek GPU sayısı (örneğin, 0,5, 1). Ayarlandığında, UDTF ayrı bir operatörde çalışır ve diğer UDF'lerle (diğer GPU UDF'leri dahil) zincirlenmez.
  gpu_type	Dize	Hayır	GPU türü. num_gpus belirtildiğinde gereklidir.

• Dönüş değeri

  flat_map veya join_lateral içinde kullanılabilecek bir DataFrameUDTFWrapper nesnesi.

In join_lateral Used in

join_lateral appends UDTF results after the original input row. By default, input rows that produce no UDTF output are removed; you can set ignore_empty=False to keep input rows with no UDTF output.

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

# Use type hint and .alias() to specify return types and column names
@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,
)

# Or use 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")),
)

In flat_map Used in

flat_map calls the UDTF on each row. The result contains only the UDTF output columns, and the original input columns are not preserved. You can pass a Python function directly to flat_map without wrapping it with the udtf decorator.

from typing import Any, Dict, Iterator, TypedDict

# Use type hint with TypedDict to specify return types and column names
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)

# Or use 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()}),
)

Text classification

• API name: ai_classify.

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

• Description: Classifies text into one of the specified labels.

• Parameters

  Parameter name	Parameter type	Required	Description
  input_col	String / Expression	Yes	The input text column name or a column expression.
  labels	List[String]	Yes	A list of classification labels, for example, ["positive", "negative", "neutral"].

  See the descriptions of other parameters at Generic invocation.

  Not

  Domain-specific AI functions such as ai_classify use a built-in system prompt. The system prompt configured on the Model Provider does not take effect.

• Return value

  A DataFrame with the category (STRING) and confidence (DOUBLE) columns appended, which contain the classification result and the confidence score.

• Example

  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")

Sentiment analysis

• API name: ai_sentiment.

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

• Description: Performs sentiment analysis on the input text.

• Parameters

  Parameter name	Parameter type	Required	Description
  input_col	String / Expression	Yes	The input text column name or a column expression.

  See the descriptions of other parameters at Generic invocation.

• Return value

  A DataFrame with the following columns appended, which contain the sentiment analysis result:

  • score (DOUBLE): The sentiment analysis score, ranging from -1.0 to 1.0.

  • label (STRING): One of "positive", "negative", or "neutral".

  • confidence (DOUBLE): Confidence score.

• Example

  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")

Information extraction

• API name: ai_extract.

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

• Description: Extracts structured information from text based on a given JSON Schema.

• Parameters

  Parameter name	Parameter type	Required	Description
  input_col	String / Expression	Yes	The input text column name or a column expression.
  schema	String	Yes	A JSON Schema string that describes the fields to extract, for example, '{"name":"STRING", "phone":"STRING"}'.

  See the descriptions of other parameters at Generic invocation.

• Return value

  A DataFrame with the extracted_json (STRING) column appended, which contains the extracted structured information.

• Example

  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")

Text translation

• API name: 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
  )

• Description: Translates text from a source language to a target language.

• Parameters

  Parameter name	Parameter type	Required	Description
  input_col	String / Expression	Yes	The input text column name or a column expression.
  source_lang	String	Yes	Source language code, for example: "zh", "en", "auto" (auto-detected). Supported: auto, zh, en, ja, ko, fr, de, es, ru, ar, pt.
  target_lang	String	Yes	The target language code. This value cannot be "auto".

  See the descriptions of other parameters at Generic invocation.

• Return value

  A DataFrame with the translated_text (STRING) and detected_language (STRING) columns appended, which contain the translated text and the detected language.

• Example

  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")

Text summarization

• API name: ai_summarize.

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

• Description: Summarizes text to a specified maximum length.

• Parameters

  Parameter name	Parameter type	Required	Description
  input_col	String / Expression	Yes	The input text column name or a column expression.
  max_length	int	Yes	Maximum number of characters in the summary. Must be greater than 0.

  See the descriptions of other parameters at Generic invocation.

• Return value

  A DataFrame with the summary (STRING) column appended, which contains the summary text.

• Example

  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")
  

Data masking

• API name: ai_mask.

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

• Description: Masks sensitive information in text.

• Parameters

  Parameter name	Parameter type	Required	Description
  input_col	String / Expression	Yes	The input text column name or a column expression.
  entities	List[String]	Yes	A list of entity types to mask, for example, ["name", "phone"].

  See the descriptions of other parameters at Generic invocation.

• Return value

  A DataFrame with the masked_text (STRING) and detected_entities (ARRAY<STRING>) columns appended, which contain the masked text and the detected entities.

• Example

  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")

Vektör araması

• API adı: 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
  )

• Açıklama: Mevcut DataFrame'deki sorgu vektörlerini kullanarak vektör arama kaynağında en benzer kayıtları arar ve arama sonuçlarını mevcut DataFrame'e ekler.

• Parametreler

  Parametre adı	Parametre türü	Zorunlu	Açıklama
  search_source	DataFrame	Evet	Vektör arama kaynağı DataFrame'i; örneğin, read_milvus kullanılarak okunan bir vektör koleksiyonu.
  column_to_search	Dize	Evet	Arama kaynağındaki vektör sütununun adı.
  column_to_query	Dize / İfade	Evet	Mevcut DataFrame'deki sorgu vektörü sütun adı veya sütun ifadesi.
  top_k	int	Evet	Giriş başına döndürülen en benzer kayıt sayısı.
  agg	Boolean	Hayır	En iyi K sonucun tek bir dizi sütununda toplanıp toplanmayacağını belirtir. Varsayılan değer: False.
  output_columns	List[Dize] / Eşleme[Dize, Dize]	Hayır	Döndürülecek vektör arama sonucu sütunları. Bir liste sütunları seçer ve orijinal adlarını korur. agg=False iken ayrıca anahtarların sütunları seçtiği ve değerlerin çıktı takma adları olduğu bir eşleme kullanabilirsiniz. search_source üst düzey sütunlarını ve motor tarafından oluşturulan score sütununu seçebilirsiniz. Varsayılan değer: None (tüm sonuç sütunlarını seçer).
  config	Sözlük	Hayır	Çalışma zamanı vektör arama yapılandırması.
  ignore_empty	Boolean	Hayır	Arama sonucu bulunmayan giriş satırlarının atılıp atılmayacağını belirtir. Varsayılan değer: False.

• Dönüş değeri

  Orijinal sütunları ve eklenen vektör arama sonucu sütunlarını içeren bir DataFrame nesnesi. output_columns toplamda M sonuç sütunu seçiyorsa:

  • output_columns belirtmezseniz, search_source içindeki tüm sütunlar ve DOUBLE türünde bir arama puanı sütunu döndürülür. Puan sütunu varsayılan olarak score olarak adlandırılır. Puanın anlamı ve sıralama yönü bağlayıcıya ve uzaklık metriğine bağlıdır. Milvus için değeri search_metric ile birlikte değerlendirin.

  • agg=False olduğunda her giriş satırı en fazla top_k adet çıktı satırı üretir ve M sütun eklenir. Bir liste sütunları bildirilen sırayla seçer ve orijinal adlarını korur. Bir eşleme ise anahtarlarının bildirim sırasına göre sütunları seçer ve karşılık gelen değerleri çıktı takma adları olarak kullanır.

  • agg=True olduğunda arama sonucu bulunan her giriş en fazla bir çıktı satırı üretir ve search_results adlı ARRAY<ROW<...>> türünde bir sütun eklenir. Varsayılan olarak ignore_empty=False tüm giriş kayıtlarını korur ve bir girdinin arama sonucu yoksa search_results değeri NULL olur. ignore_empty=True ayarlarsanız arama sonucu olmayan girdiler sonuçta yer almaz. Her dizi öğesi, seçilen M alanı içeren bir ROW yapısıdır. Bu mod yalnızca iç alanları projekte etmek için bir listeyi destekler, eşlemeyi desteklemez. Çağrıdan sonra dış dizi sütununu yeniden adlandırmak için rename_columns("search_results", "new_name") çağrısını kullanın.

  • output_columns boş olamaz; sütun adları veya takma adlar da boş bırakılamaz veya yinelenemez. Seçilen sütunlar search_source üst düzey sütunları veya oluşturulan benzerlik sütunu olmalıdır. Dönüş sırası listenin veya eşleme anahtarlarının bildirim sırasına uyar.

  • search_source zaten score veya score0 gibi sütunlara sahipse motor oluşturulan benzerlik sütunu için score2 gibi benzersiz bir ad seçer. Bu durumda output_columns içinde gerçek sütun adını kullanın.

  • Toplu işlem olmayan modda sonuç sütunlarına çakışmayan takma adlar vermek için eşleme kullanabilirsiniz. Bir liste veya varsayılan çıktı mevcut DataFrame'in sütunlarıyla çakışıyorsa vector_search çağrısından önce çakışan sütunları rename_columns ile yeniden adlandırın. Toplu işlem modunda mevcut DataFrame'de zaten bir search_results sütunu varsa önce bu sütunu yeniden adlandırın.

  • Vektör arama kaynağının bağlayıcısı sütun budama itme desteğine sahipse motor output_columns budağını bağlayıcıya iter. İtme desteği olup olmadığından bağımsız olarak nihai sonuç yalnızca seçilen vektör arama sonucu sütunlarını içerir.

• VVR 11.8'den yükseltme notları

  VVR 11.8'de output_columns tüm çıktı sütunlarının takma adını ifade eder. VVR 11.9 ve sonraki sürümlerde ise seçilecek sonuç sütunlarını ifade eder. Örneğin, VVR 11.8 için yazılan aşağıdaki çağrı:

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

  Yükseltmeden sonra bunun yerine bir eşleme kullanın:

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

• Örnekler

  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"],
  )

  Bir liste sonuç sütunlarını projekte eder ve orijinal adlarını korur. Aynı anda takma ad belirtmek için bir eşleme kullanın:

  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",
      },
  )

  Toplu işlem modunda bir liste her dizi öğesinin alanlarını seçer ve dış dizi sütunu her zaman search_results olarak adlandırılır. Dış dizi sütununu yeniden adlandırmak için rename_columns çağrısını kullanın:

  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")

Metin gömme

• API adı: ai_embed.

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

• Açıklama: Metin için vektör gömmeleri oluşturur.

• Parametreler

  Parametre adı	Parametre türü	Zorunlu	Açıklama
  input_col	Dize / İfade	Evet	Giriş metni sütun adı veya bir sütun ifadesi.
  dimension	int	Hayır	Vektör boyutu. Varsayılan değer: 1024.

  Diğer parametrelerin açıklamaları için bkz. Genel çağrı.

• Dönüş değeri

  Oluşturulan vektörü içeren embedding (ARRAY<FLOAT>) sütunu eklenmiş bir DataFrame.

• Örnek

  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")