Python 中用户定义的函数

Python 用户定义的函数 (UDF) 使您可以在 Python 中实现标量函数,并在 SQL 查询中进行使用。Python UDF 与 SQL 和 JavaScript UDF 类似,但具有更多功能。借助 Python UDF,您可以通过 Python 软件包索引 (PyPI) 安装第三方库,还可以使用 Cloud 资源连接访问外部服务。

Python UDF 在 BigQuery 托管资源上构建和运行。

限制

  • python-3.11 是唯一受支持的运行时。
  • 您无法创建临时 Python UDF。
  • 您无法将 Python UDF 与物化视图搭配使用。
  • 调用 Python UDF 的查询的结果不会进行缓存,因为 Python UDF 的返回值始终假定为具有非确定性。
  • INFORMATION_SCHEMA 视图中未完全支持 Python UDF。
  • 您无法使用 Routine API 创建或更新 Python UDF。
  • 不支持 VPC Service Controls
  • 不支持客户管理的加密密钥 (CMEK)
  • 不支持以下数据类型:JSONRANGEINTERVALGEOGRAPHY
  • 运行 Python UDF 的容器最多只能配置 2 个 vCPU 和 8 Gi

所需 IAM 角色

所需的 IAM 角色取决于您是 Python UDF 所有者还是 Python UDF 用户。Python UDF 所有者通常会创建或更新 UDF。Python UDF 用户会调用他人创建的 UDF。

如果您创建或运行引用 Cloud 资源连接的 Python UDF,还需要其他角色。

UDF 所有者

如果您要创建或更新 Python UDF,则应被授予针对相应资源的以下预定义 IAM 角色:

角色 所需权限 资源
BigQuery Data Editor (roles/bigquery.dataEditor)
  • bigquery.routines.create,用于使用 CREATE FUNCTION 语句创建 Python UDF。
  • bigquery.routines.update,用于使用 CREATE FUNCTION 语句更新 Python UDF。
在其中创建或更新 Python UDF 的数据集。
BigQuery Job User (roles/bigquery.jobUser)
  • bigquery.jobs.create,用于运行 CREATE FUNCTION 语句查询作业。
在其中运行 CREATE FUNCTION 语句的项目。
BigQuery Connection Admin (roles/bigquery.connectionAdmin)
  • bigquery.connections.create,仅在创建新的 Cloud 资源连接时才需要。
  • bigquery.connections.delegate,用于在 CREATE FUNCTION 语句中使用连接。
您向其授予对外部资源的访问权限的连接。仅当您的 UDF 使用 WITH CONNECTION 子句访问外部服务时,才需要此连接。

UDF 用户

如果您要调用 Python UDF,则应被授予针对相应资源的以下预定义 IAM 角色:

角色 所需权限 资源
BigQuery User (roles/bigquery.user) bigquery.jobs.create,用于运行引用 UDF 的查询作业。 在其中运行调用 Python UDF 的查询作业的项目。
BigQuery Data Viewer (roles/bigquery.dataViewer) bigquery.routines.get,用于运行他人创建的 UDF。 在其中存储 Python UDF 的数据集。
BigQuery Connection User (roles/bigquery.connectionUser) bigquery.connections.use,用于运行引用 Cloud 资源连接的 Python UDF。 Python UDF 引用的 Cloud 资源连接。仅当您的 UDF 引用连接时,才需要此连接。

如需详细了解 BigQuery 中的角色,请参阅预定义的 IAM 角色

创建永久性 Python UDF

创建 Python UDF 时,请遵循以下规则:

  • Python UDF 的主体必须是带英文引号的字符串字面量,用于表示 Python 代码。如需详细了解带英文引号的字符串字面量,请参阅带英文引号的字面量的格式

  • Python UDF 的主体必须包含一个 Python 函数,该函数在 Python UDF 选项列表的 entry_point 参数中使用。

  • 需要在 runtime_version 选项中指定 Python 运行时版本。唯一受支持的 Python 运行时版本是 python-3.11。如需查看可用选项的完整列表,请参阅 CREATE FUNCTION 语句的函数选项列表

如需创建永久性 Python UDF,请使用 CREATE FUNCTION 语句,并且不带 TEMPTEMPORARY 关键字。如需删除永久性 Python UDF,请使用 DROP FUNCTION 语句。

当您使用 CREATE FUNCTION 语句创建 Python UDF 时,BigQuery 会创建或更新基于基础映像的容器映像。容器使用您的代码和任何指定的软件包依赖项,基于基础映像构建而成。创建容器是一个长时间运行的过程。运行 CREATE FUNCTION 语句后的第一个查询可能会自动等待映像完成。在没有任何外部依赖项的情况下,容器映像通常应在不到一分钟的时间内创建完成。

示例

如需查看创建永久性 Python UDF 的示例,请选择以下选项之一:

控制台

以下示例创建一个名为 multiplyInputs 的永久性 Python UDF,并从 SELECT 语句中调用该 UDF:

  1. 转到 BigQuery 页面。

    转到 BigQuery

  2. 在查询编辑器中,输入以下 CREATE FUNCTION 语句:

    CREATE FUNCTION `PROJECT_ID.DATASET_ID`.multiplyInputs(x FLOAT64, y FLOAT64)
    RETURNS FLOAT64
    LANGUAGE python
    OPTIONS(runtime_version="python-3.11", entry_point="multiply")
    AS r'''
    
    def multiply(x, y):
      return x * y
    
    ''';
    
    -- Call the Python UDF.
    WITH numbers AS
      (SELECT 1 AS x, 5 as y
      UNION ALL
      SELECT 2 AS x, 10 as y
      UNION ALL
      SELECT 3 as x, 15 as y)
    SELECT x, y,
    `PROJECT_ID.DATASET_ID`.multiplyInputs(x, y) AS product
    FROM numbers;

    替换 PROJECT_IDDATASET_ID 替换为您的项目 ID 和数据集 ID。

  3. 点击  运行

    此示例生成以下输出:

    +-----+-----+--------------+
    | x   | y   | product      |
    +-----+-----+--------------+
    | 1   | 5   |  5.0         |
    | 2   | 10  | 20.0         |
    | 3   | 15  | 45.0         |
    +-----+-----+--------------+
    

BigQuery DataFrame

以下示例使用 BigQuery DataFrames 将自定义函数转换为 Python UDF:

import bigframes.pandas as bpd

# Set BigQuery DataFrames options
bpd.options.bigquery.project = your_gcp_project_id
bpd.options.bigquery.location = "US"

# BigQuery DataFrames gives you the ability to turn your custom functions
# into a BigQuery Python UDF. One can find more details about the usage and
# the requirements via `help` command.
help(bpd.udf)

# Read a table and inspect the column of interest.
df = bpd.read_gbq("bigquery-public-data.ml_datasets.penguins")
df["body_mass_g"].peek(10)

# Define a custom function, and specify the intent to turn it into a
# BigQuery Python UDF. Let's try a `pandas`-like use case in which we want
# to apply a user defined function to every value in a `Series`, more
# specifically bucketize the `body_mass_g` value of the penguins, which is a
# real number, into a category, which is a string.
@bpd.udf(
    dataset=your_bq_dataset_id,
    name=your_bq_routine_id,
)
def get_bucket(num: float) -> str:
    if not num:
        return "NA"
    boundary = 4000
    return "at_or_above_4000" if num >= boundary else "below_4000"

# Then we can apply the udf on the `Series` of interest via
# `apply` API and store the result in a new column in the DataFrame.
df = df.assign(body_mass_bucket=df["body_mass_g"].apply(get_bucket))

# This will add a new column `body_mass_bucket` in the DataFrame. You can
# preview the original value and the bucketized value side by side.
df[["body_mass_g", "body_mass_bucket"]].peek(10)

# The above operation was possible by doing all the computation on the
# cloud through an underlying BigQuery Python UDF that was created to
# support the user's operations in the Python code.

# The BigQuery Python UDF created to support the BigQuery DataFrames
# udf can be located via a property `bigframes_bigquery_function`
# set in the udf object.
print(f"Created BQ Python UDF: {get_bucket.bigframes_bigquery_function}")

# If you have already defined a custom function in BigQuery, either via the
# BigQuery Google Cloud Console or with the `udf` decorator,
# or otherwise, you may use it with BigQuery DataFrames with the
# `read_gbq_function` method. More details are available via the `help`
# command.
help(bpd.read_gbq_function)

existing_get_bucket_bq_udf = get_bucket.bigframes_bigquery_function

# Here is an example of using `read_gbq_function` to load an existing
# BigQuery Python UDF.
df = bpd.read_gbq("bigquery-public-data.ml_datasets.penguins")
get_bucket_function = bpd.read_gbq_function(existing_get_bucket_bq_udf)

df = df.assign(body_mass_bucket=df["body_mass_g"].apply(get_bucket_function))
df.peek(10)

# Let's continue trying other potential use cases of udf. Let's say we
# consider the `species`, `island` and `sex` of the penguins sensitive
# information and want to redact that by replacing with their hash code
# instead. Let's define another scalar custom function and decorate it
# as a udf. The custom function in this example has external package
# dependency, which can be specified via `packages` parameter.
@bpd.udf(
    dataset=your_bq_dataset_id,
    name=your_bq_routine_id,
    packages=["cryptography"],
)
def get_hash(input: str) -> str:
    from cryptography.fernet import Fernet

    # handle missing value
    if input is None:
        input = ""

    key = Fernet.generate_key()
    f = Fernet(key)
    return f.encrypt(input.encode()).decode()

# We can use this udf in another `pandas`-like API `map` that
# can be applied on a DataFrame
df_redacted = df[["species", "island", "sex"]].map(get_hash)
df_redacted.peek(10)

# If the BigQuery routine is no longer needed, we can clean it up
# to free up any cloud quota
session = bpd.get_global_session()
session.bqclient.delete_routine(f"{your_bq_dataset_id}.{your_bq_routine_id}")

创建向量化 Python UDF

您可以使用向量化技术来实现 Python UDF,以处理一批行而不是单个行。向量化可以提高查询性能。

如需控制批处理行为,请使用 CREATE OR REPLACE FUNCTION 选项列表中的 max_batching_rows 选项指定每个批次中的最大行数。如果您指定 max_batching_rows,BigQuery 会确定批次中的行数(不会超过 max_batching_rows 限制)。如果未指定 max_batching_rows,系统会自动确定要进行批处理的行数。

向量化 Python UDF 具有必须进行注解的单个 pandas.DataFrame 参数。pandas.DataFrame 参数的列数与 CREATE FUNCTION 语句中定义的 Python UDF 参数相同。pandas.DataFrame 参数中的列名称与 UDF 的参数名称相同。

您的函数需要返回 pandas.Series 或单列 pandas.DataFrame,且行数与输入相同。

以下示例创建一个名为 multiplyInputs 的向量化 Python UDF,其中包含两个参数 xy

  1. 转到 BigQuery 页面。

    转到 BigQuery

  2. 在查询编辑器中,输入以下 CREATE FUNCTION 语句:

    CREATE FUNCTION `PROJECT_ID.DATASET_ID`.multiplyVectorized(x FLOAT64, y FLOAT64)
    RETURNS FLOAT64
    LANGUAGE python
    OPTIONS(runtime_version="python-3.11", entry_point="vectorized_multiply")
    AS r'''
    import pandas as pd
    
    def vectorized_multiply(df: pd.DataFrame):
      return df['x'] * df['y']
    
    ''';

    替换 PROJECT_IDDATASET_ID 替换为您的项目 ID 和数据集 ID。

    调用 UDF 的方式与上一个示例相同。

  3. 点击  运行

支持的 Python UDF 数据类型

下表定义了 BigQuery 数据类型、Python 数据类型和 Pandas 数据类型之间的映射:

BigQuery 数据类型 标准 UDF 使用的 Python 内置数据类型 向量化 UDF 使用的 Pandas 数据类型 向量化 UDF 中用于 ARRAY 和 STRUCT 的 PyArrow 数据类型
BOOL bool BooleanDtype DataType(bool)
INT64 int Int64Dtype DataType(int64)
FLOAT64 float FloatDtype DataType(double)
STRING str StringDtype DataType(string)
BYTES bytes binary[pyarrow] DataType(binary)
TIMESTAMP

函数参数:datetime.datetime(设置了世界协调时间 [UTC] 时区)

函数返回值:datetime.datetime(设置了任意时区)

函数参数:timestamp[us, tz=UTC][pyarrow]

函数返回值:timestamp[us, tz=*][pyarrow]\(any timezone\)

TimestampType(timestamp[us]),带有时区
DATE datetime.date date32[pyarrow] DataType(date32[day])
TIME datetime.time time64[pyarrow] Time64Type(time64[us])
DATETIME datetime.datetime(不带时区) timestamp[us][pyarrow] TimestampType(timestamp[us]),不带时区
ARRAY list list<...>[pyarrow],其中元素数据类型为 pandas.ArrowDtype ListType
STRUCT dict struct<...>[pyarrow],其中字段数据类型为 pandas.ArrowDtype StructType

支持的运行时版本

BigQuery Python UDF 支持 python-3.11 运行时。此 Python 版本包含一些额外的预安装软件包。对于系统库,请检查运行时基础映像。

运行时版本 Python 版本 包含 运行时基础映像
python-3.11 Python 3.11 numpy 1.26.3
pyarrow 14.0.2
pandas 2.1.4
python-dateutil 2.8.2
google-22-full/python311

使用第三方软件包

您可以通过 CREATE FUNCTION 选项列表来使用 Python 标准库和预安装软件包提供的模块以外的模块。您可以通过 Python 软件包索引 (PyPI) 安装软件包,也可以从 Cloud Storage 导入 Python 文件。

通过 Python 软件包索引安装软件包

安装软件包时,您必须提供软件包名称,还可以选择使用 Python 软件包版本说明符提供软件包版本。如果软件包在运行时中,则使用该软件包,除非在 CREATE FUNCTION 选项列表中指定了特定版本。如果未指定软件包版本,并且软件包不在运行时中,则使用最新可用版本。仅支持采用 wheels 二进制格式的软件包。

以下示例展示了如何创建使用 CREATE OR REPLACE FUNCTION 选项列表安装 scipy 软件包的 Python UDF:

  1. 转到 BigQuery 页面。

    转到 BigQuery

  2. 在查询编辑器中,输入以下 CREATE FUNCTION 语句:

    CREATE FUNCTION `PROJECT_ID.DATASET_ID`.area(radius FLOAT64)
    RETURNS FLOAT64 LANGUAGE python
    OPTIONS (entry_point='area_handler', runtime_version='python-3.11', packages=['scipy==1.15.3'])
    AS r"""
    import scipy
    
    def area_handler(radius):
      return scipy.constants.pi*radius*radius
    """;
    
    SELECT `PROJECT_ID.DATASET_ID`.area(4.5);

    替换 PROJECT_IDDATASET_ID 替换为您的项目 ID 和数据集 ID。

  3. 点击  运行

将其他 Python 文件作为库导入

您可以通过从 Cloud Storage 导入 Python 文件,使用函数选项列表来扩展 Python UDF。