Realtime Compute for Apache Flink:Python DataFrame API 参考

Classification de texte

• Nom de l'API : ai_classify.

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

• Description : Classe le texte dans l'une des étiquettes spécifiées.

• Paramètres

  Nom du paramètre	Type du paramètre	Obligatoire	Description
  input_col	String / Expression	Oui	Nom de la colonne de texte d'entrée ou expression de colonne.
  labels	List[String]	Oui	Liste d'étiquettes de classification, par exemple ["positive", "negative", "neutral"].

  Consultez les descriptions des autres paramètres dans la section Appel générique.

  Remarque

  Les fonctions d'IA spécifiques à un domaine, telles que ai_classify, utilisent un prompt système intégré. Le prompt système configuré sur le fournisseur de modèles n'est pas pris en compte.

• Valeur de retour

  Un DataFrame auquel sont ajoutées les colonnes category (STRING) et confidence (DOUBLE), contenant respectivement le résultat de la classification et le score de confiance.

• Exemple

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

Analyse des sentiments

• Nom de l'API : ai_sentiment.

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

• Description : Effectue une analyse des sentiments sur le texte d'entrée.

• Paramètres

  Nom du paramètre	Type du paramètre	Obligatoire	Description
  input_col	String / Expression	Oui	Nom de la colonne de texte d'entrée ou expression de colonne.

  Consultez les descriptions des autres paramètres dans la section Appel générique.

• Valeur de retour

  Un DataFrame auquel sont ajoutées les colonnes suivantes, contenant le résultat de l'analyse des sentiments :

  • score (DOUBLE) : Score d'analyse des sentiments, allant de -1,0 à 1,0.

  • label (STRING) : « positive », « negative » ou « neutral ».

  • confidence (DOUBLE) : Score de confiance.

• Exemple

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

Extraction d'informations

• Nom de l'API : ai_extract.

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

• Description : Extrait des informations structurées à partir d'un texte en se basant sur un schéma JSON donné.

• Paramètres

  Nom du paramètre	Type du paramètre	Obligatoire	Description
  input_col	String / Expression	Oui	Nom de la colonne de texte d'entrée ou expression de colonne.
  schema	String	Oui	Chaîne de schéma JSON décrivant les champs à extraire, par exemple '{"name":"STRING", "phone":"STRING"}'.

  Consultez les descriptions des autres paramètres dans la section Appel générique.

• Valeur de retour

  Un DataFrame auquel est ajoutée la colonne extracted_json (STRING), contenant les informations structurées extraites.

• Exemple

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

Traduction de texte

• Nom de l'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
  )

• Description : Traduit un texte d'une langue source vers une langue cible.

• Paramètres

  Nom du paramètre	Type du paramètre	Obligatoire	Description
  input_col	String / Expression	Oui	Nom de la colonne de texte d'entrée ou expression de colonne.
  source_lang	String	Oui	Code de la langue source, par exemple : "zh", "en", "auto" (détection automatique). Langues prises en charge : auto, zh, en, ja, ko, fr, de, es, ru, ar, pt.
  target_lang	String	Oui	Code de la langue cible. Cette valeur ne peut pas être "auto".

  Consultez les descriptions des autres paramètres dans la section Appel générique.

• Valeur de retour

  Un DataFrame auquel sont ajoutées les colonnes translated_text (STRING) et detected_language (STRING), contenant respectivement le texte traduit et la langue détectée.

• Exemple

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

Résumé de texte

• Nom de l'API : ai_summarize.

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

• Description : Résume un texte jusqu'à une longueur maximale spécifiée.

• Paramètres

  Nom du paramètre	Type du paramètre	Obligatoire	Description
  input_col	String / Expression	Oui	Nom de la colonne de texte d'entrée ou expression de colonne.
  max_length	int	Oui	Nombre maximal de caractères dans le résumé. Doit être supérieur à 0.

  Consultez les descriptions des autres paramètres dans la section Appel générique.

• Valeur de retour

  Un DataFrame auquel est ajoutée la colonne summary (STRING), contenant le texte du résumé.

• Exemple

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

Masquage de données

• Nom de l'API : ai_mask.

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

• Description : Masque les informations sensibles dans un texte.

• Paramètres

  Nom du paramètre	Type du paramètre	Obligatoire	Description
  input_col	String / Expression	Oui	Nom de la colonne de texte d'entrée ou expression de colonne.
  entities	List[String]	Oui	Liste des types d'entités à masquer, par exemple ["name", "phone"].

  Consultez les descriptions des autres paramètres dans la section Appel générique.

• Valeur de retour

  Un DataFrame auquel sont ajoutées les colonnes masked_text (STRING) et detected_entities (ARRAY<STRING>), contenant respectivement le texte masqué et les entités détectées.

• Exemple

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

Recherche vectorielle

• Nom de l'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
  )

• Description : Recherche les enregistrements les plus similaires dans la source de recherche vectorielle en utilisant les vecteurs de requête du DataFrame actuel, et ajoute les résultats de la recherche au DataFrame actuel.

• Paramètres

  Nom du paramètre	Type de paramètre	Obligatoire	Description
  search_source	DataFrame	Oui	Le DataFrame source de recherche vectorielle, par exemple une collection vectorielle lue à l'aide de read_milvus.
  column_to_search	String	Oui	Le nom de la colonne vectorielle dans la source de recherche.
  column_to_query	String / Expression	Oui	Le nom de la colonne de vecteur de requête ou l'expression de colonne dans le DataFrame actuel.
  top_k	int	Oui	Nombre d'enregistrements les plus similaires renvoyés par enregistrement d'entrée.
  agg	Boolean	Non	Indique s'il faut agréger les K meilleurs résultats en une seule colonne de tableau. Valeur par défaut : False.
  output_columns	List[String] / Mapping[String, String]	Non	Les colonnes de résultat de recherche vectorielle à renvoyer. Une liste sélectionne les colonnes et conserve leurs noms d'origine. Avec agg=False, vous pouvez également utiliser un mapping, où les clés sélectionnent les colonnes et les valeurs deviennent des alias de sortie. Vous pouvez sélectionner les colonnes de premier niveau de search_source et la colonne score générée par le moteur. Valeur par défaut : None, qui sélectionne toutes les colonnes de résultat.
  config	Dict	Non	La configuration d'exécution de la recherche vectorielle.
  ignore_empty	Boolean	Non	Indique s'il faut supprimer les lignes d'entrée qui n'ont aucun résultat de recherche. Valeur par défaut : False.

• Valeur de retour

  Un objet DataFrame contenant les colonnes d'origine et les colonnes de résultat de recherche vectorielle ajoutées. Supposons que output_columns sélectionne M colonnes de résultat au total :

  • Si vous ne spécifiez pas output_columns, toutes les colonnes de search_source et une colonne de score de recherche de type DOUBLE sont renvoyées. Par défaut, la colonne de score est nommée score. La signification et l'ordre de tri du score dépendent du connecteur et de sa métrique de distance. Pour Milvus, interprétez la valeur conjointement avec search_metric.

  • Lorsque agg=False, chaque ligne d'entrée produit au plus top_k lignes de sortie avec M colonnes ajoutées. Une liste sélectionne les colonnes dans l'ordre déclaré et conserve leurs noms d'origine. Un mapping sélectionne les colonnes dans l'ordre de déclaration de ses clés et utilise les valeurs correspondantes comme alias de sortie.

  • Lorsque agg=True, chaque entrée ayant des résultats de recherche produit au plus une ligne de sortie, avec une colonne search_results de type ARRAY<ROW<...>> ajoutée. Par défaut, ignore_empty=False conserve tous les enregistrements d'entrée, et search_results est NULL lorsqu'une entrée n'a aucun résultat de recherche. Si vous définissez ignore_empty=True, les entrées sans résultats de recherche n'apparaissent pas dans le résultat. Chaque élément du tableau est une ROW contenant les M champs sélectionnés. Ce mode prend uniquement en charge une liste pour projeter les champs internes, et non un mapping. Après l'appel, renommez la colonne de tableau externe en appelant rename_columns("search_results", "new_name").

  • output_columns ne peut pas être vide, et ni les noms de colonnes ni les alias ne peuvent être vides ou dupliqués. Les colonnes sélectionnées doivent être des colonnes de premier niveau de search_source ou la colonne de similarité générée. L'ordre renvoyé correspond à l'ordre de déclaration de la liste ou des clés du mapping.

  • Si search_source possède déjà des colonnes telles que score ou score0, le moteur choisit un nom unique pour la colonne de similarité générée, tel que score2. Dans ce cas, utilisez le nom réel de la colonne dans output_columns.

  • En mode sans agrégation, vous pouvez utiliser un mapping pour attribuer aux colonnes de résultat des alias qui n'entrent pas en conflit. Si une liste ou la sortie par défaut entre en conflit avec les colonnes existantes du DataFrame actuel, renommez les colonnes conflictuelles avec rename_columns avant d'appeler vector_search. En mode avec agrégation, si le DataFrame actuel possède déjà une colonne search_results, renommez d'abord cette colonne.

  • Si le connecteur de la source de recherche vectorielle prend en charge la propagation descendante de l'élagage des colonnes, le moteur propage l'élagage de output_columns jusqu'au connecteur. Que la propagation soit prise en charge ou non, le résultat final contient uniquement les colonnes de résultat de recherche vectorielle sélectionnées.

• Notes sur la mise à niveau depuis VVR 11.8

  Dans VVR 11.8, output_columns désigne l'alias de toutes les colonnes de sortie. Dans VVR 11.9 et versions ultérieures, il désigne les colonnes de résultat à sélectionner. Par exemple, l'appel suivant écrit pour VVR 11.8 :

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

  Après la mise à niveau, utilisez plutôt un mapping :

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

• Exemples

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

  Une liste projette les colonnes de résultat et conserve leurs noms d'origine. Pour spécifier des alias simultanément, utilisez un mapping :

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

  En mode avec agrégation, une liste sélectionne les champs de chaque élément du tableau, et la colonne de tableau externe est toujours nommée search_results. Pour renommer la colonne de tableau externe, appelez 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")

Intégration de texte

• Nom de l'API : ai_embed.

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

• Description : Génère des intégrations vectorielles pour le texte.

• Paramètres

  Nom du paramètre	Type de paramètre	Obligatoire	Description
  input_col	String / Expression	Oui	Le nom de la colonne de texte d'entrée ou une expression de colonne.
  dimension	int	Non	La dimension du vecteur. Valeur par défaut : 1024.

  Consultez les descriptions des autres paramètres dans Invocation générique.

• Valeur de retour

  Un DataFrame avec la colonne embedding (ARRAY<FLOAT>) ajoutée, qui contient le vecteur généré.

• Exemple

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

Opérateurs multimodaux

L'API DataFrame vous permet d'appeler des opérateurs multimodaux pour lire, traiter et analyser des données multimodales telles que des images, de l'audio et de la vidéo. Pour plus d'informations, consultez Opérateurs multimodaux. Pour la documentation de l'API, consultez Expressions multimodales.

Environnement et configuration