Python DataFrame API 参考

表达式辅助函数

数据读入

数据输出

udf

• 接口名称：udf。

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

• 功能描述：将 Python 函数注册为 DataFrame 可用的标量用户自定义函数。支持普通函数、async 函数、ScalarFunction / AsyncScalarFunction 子类及可调用类。

• 入参

  参数名称	参数类型	是否必填	描述
  func	Callable / Class	否	要包装的 Python 函数、ScalarFunction 实例/子类或可调用类。若省略则返回装饰器。
  return_dtype	DataType / str / type	否	返回类型。可以是 DataType 实例（如 DataType.int64()）、Python 类型（如 int）或 SQL 类型字符串（如 'BIGINT'）。若省略则从函数返回类型提示自动推断。
  deterministic	Boolean	否	函数是否是确定性的。默认值为 True。
  name	String	否	UDF 的名称。若不指定则使用函数名。
  func_type	String	否	执行格式，可选 "general"、"pandas"、"arrow"。若不指定则根据参数类型提示自动检测。
  concurrency	int	否	UDF 算子的并行度。不同 concurrency 值的 UDF 会被拆分到独立算子。
  batch_size	int	否	每批最大元素数，仅适用于批处理 UDF（pandas / arrow 模式）。

• 返回值

  DataFrameUDFWrapper 对象，可通过 udf_func(col("a"), col("b")) 的方式在 with_column、with_columns、map、map_batches等操作中使用。

• 示例：请参见用户自定义函数。

Provider 配置

使用 AI 函数前，需先注册模型服务商（Provider）。

支持的 Provider 类型：

set_provider

• 接口名称：set_provider。

  set_provider(
      name_or_provider,
      provider=None,
      **options
  )

• 功能描述：注册一个全局模型 Provider 配置。支持三种调用方式：传入 Provider 实例、传入名称 + Provider 实例、传入名称 + 关键字参数。

• 入参

  参数名称	参数类型	是否必填	描述
  name_or_provider	Provider / String	是	Provider 实例（自动以其默认名称注册）或自定义名称字符串。
  provider	Provider	否	Provider 实例，仅在第一个参数为名称字符串时使用。
  **options	key=value	否	Provider 配置项（如 endpoint、api_key），仅在第一个参数为名称字符串且未传入 provider 时使用，将创建 GenericProvider。

• 返回值

  无。

• 示例

  import pyflink.dataframe as pf
  
  # Option 1: pass a Provider instance directly
  pf.set_provider(pf.OpenAICompatProvider(
      endpoint="https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
      api_key="sk-..."
  ))
  
  # Option 2: custom name + Provider instance (register multiple)
  pf.set_provider("chat", pf.OpenAICompatProvider(
      endpoint="https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
      api_key="sk-..."
  ))
  pf.set_provider("embedding", pf.OpenAICompatProvider(
      endpoint="https://dashscope.aliyuncs.com/compatible-mode/v1/embeddings",
      api_key="sk-..."
  ))
  
  # Option 3: name + keyword arguments (creates GenericProvider)
  pf.set_provider("openai-compat",
      endpoint="https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
      api_key="sk-..."
  )

通用调用

• 接口名称：predict。

  DataFrame.llm.predict(
      *input_cols,
      provider=None,
      model=None,
      output_type=None,
      config=None
  )
  

• 功能描述：通用模型推理。将指定输入列发送给模型，返回模型输出列追加到 DataFrame 中。支持自定义输出 schema。

• 入参

  参数名称	参数类型	是否必填	描述
  *input_cols	String	是	作为模型输入的列名，可传入多个。
  provider	String	否	Provider 名称。若不指定则使用默认 Provider；若无 Provider 配置，则 model 被视为 Catalog Model 名称。
  model	String	否	模型名称（如 "qwen-plus"）或 Catalog Model 名称。
  output_type	Dict	否	输出列 schema，格式为 {列名: 类型}，类型可以是 SQL 类型字符串或 DataType 对象。默认值为 {"output": "STRING"}。
  config	Dict	否	运行时配置选项。

• 返回值

  DataFrame 对象，包含原始列和模型输出列。

• 示例

  import pyflink.dataframe as pf
  
  pf.set_provider(pf.OpenAICompatProvider(
      endpoint="https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
      api_key="sk-..."
  ))
  
  df = pf.from_dict({"question": ["What is Flink?", "What is stream processing?"]})
  
  # default output column: output (STRING)
  df = df.llm.predict("question", provider="chat", model="qwen-plus")
  
  # custom output schema
  df = df.llm.predict("question", provider="chat", model="qwen-plus",
                      output_type={"answer": "STRING", "score": "DOUBLE"})
  

文本类

向量类

修改运行参数

• 接口名称：DataFrameConfig.set。

  DataFrameConfig.set(
      key, value
  )

• 功能描述：修改 DataFrame API 作业的运行参数。您可按照下方示例的方法配置 Python 作业参数和 Table API 参数，配置的参数会应用到默认的 TableEnvironment 上，且会覆盖作业运行参数配置中的同名参数。

• 入参

  参数名称	参数类型	是否必填	描述
  key	String	是	参数名。
  value	String	是	参数值。

• 返回值

  无。

• 示例

  import pyflink.dataframe as pf
  
  pf.config.set("python.fn-execution.arrow.batch.size", "256")
  pf.config.set("table.exec.async-scalar.buffer-capacity", "10")
  pf.config.set("table.exec.async-lookup.timeout", "1 min")

  说明

  建议您在编写作业时，先配置参数再定义作业逻辑，以确保参数按预期生效。