使用 BigQuery DataFrames

BigQuery DataFrames 提供由 BigQuery 引擎提供支持的 Pythonic DataFrame 和机器学习 (ML) API。BigQuery DataFrames 是一个开源软件包。您可以运行 pip install --upgrade bigframes 来安装最新版本。

BigQuery DataFrames 提供两个库：

  • bigframes.pandas：提供与 Pandas 兼容的 API 以用于分析。

  • bigframes.ml：提供类似于 scikit-learn 的 API 以用于机器学习 (ML)。

所需权限

选项

安装完成后，您需要指定要在其中使用 BigQuery DataFrames 的位置和项目。您可以通过以下方式在笔记本中定义位置和项目： 
import bigframes.pandas as bpd

PROJECT_ID = "bigframes-dec"  # @param {type:"string"}
REGION = "US"  # @param {type:"string"}

# Set BigQuery DataFrames options
# Note: The project option is not required in all environments.
# On BigQuery Studio, the project ID is automatically detected.
bpd.options.bigquery.project = PROJECT_ID

# Note: The location option is not required.
# It defaults to the location of the first table or query
# passed to read_gbq(). For APIs where a location can't be
# auto-detected, the location defaults to the "US" location.
bpd.options.bigquery.location = REGION

如果未设置 bf.options.bigquery.project，则系统会使用 $GOOGLE_CLOUD_PROJECT 环境变量，该变量在提供 BigQuery Studio 和 Vertex AI 笔记本的笔记本运行时中设置。

数据处理位置

BigQuery DataFrames 旨在实现缩放功能，通过在 BigQuery 服务上保留数据并进行处理来实现。但是，您可以通过对 DataFrame 或 Series 对象调用 .to_pandas() 将数据放入客户端机器的内存中。如果您选择这样做，则应遵循客户端机器的内存限制。

会议位置

BigQuery DataFrames 使用本地会话对象在内部管理元数据。此会话与某个位置关联。BigQuery DataFrames 使用 US 多区域作为默认位置，但您可以使用 session_options.location 设置其他位置。会话中的每个查询都在创建会话的位置执行。如果用户以 read_gbq/read_gbq_table/read_gbq_query() 开头并直接或通过 SQL 语句指定表，BigQuery DataFrames 会使用表的位置自动填充 bf.options.bigquery.location。

如果要重置所创建的 DataFrame 或 Series 对象的位置，您可以通过执行 bigframes.pandas.close_session() 来关闭会话。之后，您可以重复使用 bigframes.pandas.options.bigquery.location 来指定其他位置。

如果您所查询的数据集不在 US 多区域中，read_gbq() 会要求您指定位置。如果您尝试从其他位置读取表，则会收到 NotFound 异常。

数据类型

BigQuery DataFrames 支持以下 numpy 和 pandas dtype：

BigQuery BigQuery DataFrames 和 Pandas
ARRAY pandas.ArrowDtype(pa.list_())
BOOL pandas.BooleanDtype()
DATE pandas.ArrowDtype(pa.date32())
DATETIME pandas.ArrowDtype(pa.timestamp("us"))
FLOAT64 pandas.Float64Dtype()
GEOGRAPHY

geopandas.array.GeometryDtype()

受 to_pandas() only 支持
INT64 pandas.Int64Dtype()
STRING pandas.StringDtype(storage="pyarrow")
STRUCT pandas.ArrowDtype(pa.struct())
TIME pandas.ArrowDtype(pa.time64("us"))
TIMESTAMP pandas.ArrowDtype(pa.timestamp("us", tz="UTC"))

BigQuery DataFrames 不支持以下 BigQuery 数据类型：

  • NUMERIC

  • BIGNUMERIC

  • INTERVAL

  • RANGE

  • JSON

所有其他 BigQuery 数据类型都显示为对象类型。

部分排序模式

BigQuery DataFrames 提供了排序模式功能。将 ordering_mode 设置为 partial 可生成更高效的查询。

partial 排序模式与默认的 strict 模式形成对比，后者会对所有行创建总排序。总排序可使用 DataFrame.iloc 属性提供基于顺序的行访问，从而使 BigQuery DataFrames 与 pandas 更兼容。不过，总排序和针对该排序的默认顺序索引意味着，除非将列过滤条件和行过滤条件作为参数应用于 read_gbq 和 read_gbq_table 函数，否则这些过滤条件都不会减少扫描的字节数量。为了对 DataFrame 中的所有行提供总排序，BigQuery DataFrames 会创建所有行的哈希。这可能会导致进行忽略行和列过滤条件的完整数据扫描。

将 ordering_mode 属性设为 partial 后，BigQuery DataFrames 便不会对所有行生成总排序。部分排序模式还会关闭需要对所有行进行总排序的功能，例如 DataFrame.iloc 属性。部分排序模式会将 DefaultIndexKind 设置为 null 索引，而不是针对排列的顺序索引。

在将 ordering_mode 设置为 partial 的情况下过滤 DataFrame 时，BigQuery DataFrames 无需再计算顺序索引中缺少哪些行，因此生成的查询速度更快且效率更高。BigQuery DataFrames API 仍然类似于 pandas，就像使用严格排序模式时的默认体验一样。不过，部分排序模式与常见的 pandas 行为不同，例如，部分排序模式不会按索引执行隐式联接。

对于部分排序模式和严格排序模式，您都需要为所使用的 BigQuery 资源付费。不过，在处理大型聚簇表和/或分区表时，使用部分排序模式可以降低成本，因为针对聚簇列和分区列的行过滤条件可以减少处理的字节数量。

用法

如需使用部分排序，请先将 ordering_mode 设置为 partial，然后再使用 BigQuery DataFrames 执行任何其他操作，如以下代码示例所示：

import bigframes.pandas as bdp

bpd.options.bigquery.ordering_mode = "partial"

由于部分排序模式没有顺序索引，因此不相关的 BigQuery DataFrames 不会隐式联接。相反，您必须明确调用 DataFrame.merge 方法，才能联接源自不同表表达式的两个 BigQuery DataFrame。

Series.unique() 和 Series.drop_duplicates() 功能与部分排序模式不兼容。请改用 groupby 方法，采用以下方式查找唯一值：

unique_col = df.groupby(["column"], as_index=False).size().drop(columns="size")

使用部分排序模式时，无法保证 DataFrame.head(n) 和 Series.head(n) 函数的输出在所有调用中具有幂等性。如需下载任意小型数据样本，请使用 DataFrame.peek() 或 Series.peek() 方法。

如需查看使用 ordering_mode = "partial" 属性的详细教程，请参阅此 BigQuery DataFrames 笔记本，其中演示了如何使用部分排序模式。

问题排查

需要排序错误

某些功能需要排序，例如 DataFrame.head() 和 DataFrame.iloc 函数。如需查看需要排序的功能列表，请参阅支持的 Pandas API 中的需要排序列。

如果没有进行对象排序，操作会失败，并显示 OrderRequiredError 消息，如下所示：

OrderRequiredError: Op iloc requires an ordering. Use .sort_values or .sort_index to provide an ordering.

如错误消息所述，您可以使用 DataFrame.sort_values() 方法提供排序，以便按一个或多个列进行排序。其他操作（例如 DataFrame.groupby() 操作）会按键隐式提供针对组的总排序。

如果无法确定排序是针对所有行的完全稳定总排序，后续操作可能会向您发出警告，并显示 AmbiguousWindowWarning 消息，如下所示：

AmbiguousWindowWarning: Window ordering may be ambiguous, this can cause unstable results.

如果您的工作负载可以适应非确定性结果，或者您可以手动验证您提供的排序是否为总排序，则可以通过以下方式过滤 AmbiguousWindowWarning 消息：

import warnings

import bigframes.exceptions

warnings.simplefilter("ignore", category=bigframes.exceptions.AmbiguousWindowWarning)

Null 索引错误

某些功能（例如 DataFrame.unstack() 和 Series.interpolate() 属性）需要索引。如需查看需要索引的功能列表，请参阅支持的 pandas API 中的需要索引列。

如果您在部分排序模式下使用需要索引的操作，该操作会引发 NullIndexError 消息，如下所示：

NullIndexError: DataFrame cannot perform interpolate as it has no index. Set an index using set_index.

如错误消息所述，您可以使用 DataFrame.set_index() 方法提供索引，以便按一个或多个列进行排序。除非设置 as_index=False 参数，否则其他操作（例如 DataFrame.groupby() 操作）会按键隐式提供针对组的索引。

使用 bigframes.pandas 库

bigframes.pandas 库提供类似于 pandas 的 API，可用于分析和操作 BigQuery 中的数据。bigframes.pandas API 可以伸缩，支持处理 TB 级的 BigQuery 数据，并使用 BigQuery 查询引擎执行计算。
bigframes.pandas API 提供以下功能：

输入和输出

您可以从各种来源访问数据，包括本地 CSV 文件、Cloud Storage 文件、pandas DataFrame、BigQuery 模型和 BigQuery 函数，并将其加载到 BigQuery DataFrames DataFrame 中。您还可以从 BigQuery DataFrames 创建 BigQuery 表。

数据操作

您可以使用 Python（而不是 SQL）进行开发。您可以使用 Python 开发所有 BigQuery 数据操作，无需在语言之间切换并尝试以文本字符串形式捕获 SQL 语句。bigframes.pandas API 提供超过 750 个 pandas 函数。

Python 生态系统和可视化

bigframes.pandas API 是完整的 Python 工具生态系统的网关。此 API 支持高级统计操作，您可以直观呈现从 BigQuery DataFrames 生成的聚合。您还可以使用内置的采样操作从 BigQuery DataFrames DataFrame 切换到 pandas DataFrame。

自定义 Python 函数

借助 BigQuery DataFrames，您可以将自定义标量函数转换为 BigQuery 远程函数。在 BigQuery DataFrames 中创建远程函数会创建以下内容：

  1. Cloud Run Functions（第 2 代）函数。

  2. BigQuery 连接。默认情况下，系统会使用名为 bigframes-default-connection 的连接。如果您愿意，可以使用预配置的 BigQuery 连接，在这种情况下会跳过连接创建。

    系统会向默认连接的服务账号授予 Cloud Run Invoker (roles/run.invoker) IAM 角色。

  3. 通过 BigQuery 连接 (2) 使用 Cloud Functions 函数 (1) 的 BigQuery 远程函数。

如需查看示例，请参阅创建远程函数。

系统会使用您在自定义函数定义中提供的名称在 BigQuery DataFrames 会话所在的相同位置创建 BigQuery 连接。如需查看和管理连接，请执行以下操作：

  1. 在 Google Cloud 控制台中转至 BigQuery。

  2. 选择您在其中创建了远程函数的项目。

  3. 在“探索器”窗格中，展开该项目，然后展开“外部连接”。

BigQuery 远程函数在您指定的数据集中创建，或在匿名数据集（一种隐藏数据集类型）中创建。如果您在创建远程函数期间没有为其设置名称，BigQuery DataFrames 会应用以 bigframes 前缀开头的默认名称。如需查看和管理在用户指定的数据集中创建的远程函数，请执行以下操作：

  1. 在 Google Cloud 控制台中转至 BigQuery。

  2. 选择您在其中创建了远程函数的项目。

  3. 在“探索器”窗格中，展开该项目，展开您在其中创建远程函数的数据集，然后展开“例程”。

如需查看和管理 Cloud Run functions 函数，请使用函数页面，并使用项目选择器选择您在其中创建了函数的项目。为便于识别，由 BigQuery DataFrames 创建的函数的名称以 bigframes 为前缀。

您可以通过以下方式清理未命名的 BigQuery 远程函数及其关联的 Cloud Run functions 函数：

  • 对于 BigQuery DataFrames session，请使用 session.close()。
  • 对于默认的 BigQuery DataFrames 会话，请使用 bigframes.pandas.close_session()。
  • 对于具有 session_id 的过去会话，请使用 bigframes.pandas.clean_up_by_session_id(session_id)。

要求

如需使用 BigQuery DataFrames 远程函数，您必须启用以下 API：

如需使用 BigQuery DataFrames 远程函数，您必须在项目中被授予以下 IAM 角色：

限制

  • 首次创建远程函数时，大约需要 90 秒才能使用它们。

  • 笔记本中的细微更改（例如插入新单元或重命名变量）可能会导致重新创建远程函数，即使这些更改与远程函数代码无关。

  • BigQuery DataFrames 不会区分您在远程函数代码中添加的任何个人数据。远程函数代码会序列化为不透明盒子，以便部署为 Cloud Run functions 函数。

  • BigQuery DataFrames 创建的 Cloud Run functions（第 2 代）函数、BigQuery 连接和 BigQuery 远程函数仍保留在 Google Cloud 中。如果您不想保留这些资源，则必须使用适当的 Cloud Run functions 或 BigQuery 界面单独删除它们。

  • 一个项目一次最多可以有 1000 个 Cloud Run functions（第 2 代）函数。如需了解所有限制，请参阅 Cloud Run functions 配额。

bigframes.pandas 示例

以下示例展示了使用 bigframes.pandas 的常见方式。

从 BigQuery 表或查询加载数据

您可以通过以下方式从 BigQuery 表或查询创建 DataFrame：

# Create a DataFrame from a BigQuery table:
import bigframes.pandas as bpd

query_or_table = "bigquery-public-data.ml_datasets.penguins"
bq_df = bpd.read_gbq(query_or_table)

从 CSV 文件加载数据

您可以通过以下方式从本地或 Cloud Storage CSV 文件创建 DataFrame：

import bigframes.pandas as bpd

filepath_or_buffer = "gs://cloud-samples-data/bigquery/us-states/us-states.csv"
df_from_gcs = bpd.read_csv(filepath_or_buffer)
# Display the first few rows of the DataFrame:
df_from_gcs.head()

检查和操纵数据

您可以使用 bigframes.pandas 执行数据检查和计算操作。
以下代码示例展示如何使用 bigframes.pandas 检查 body_mass_g 列，计算平均值 body_mass，以及按 species 计算平均值 body_mass：

import bigframes.pandas as bpd

# Load data from BigQuery
query_or_table = "bigquery-public-data.ml_datasets.penguins"
bq_df = bpd.read_gbq(query_or_table)

# Inspect one of the columns (or series) of the DataFrame:
bq_df["body_mass_g"]

# Compute the mean of this series:
average_body_mass = bq_df["body_mass_g"].mean()
print(f"average_body_mass: {average_body_mass}")

# Find the heaviest species using the groupby operation to calculate the
# mean body_mass_g:
(
    bq_df["body_mass_g"]
    .groupby(by=bq_df["species"])
    .mean()
    .sort_values(ascending=False)
    .head(10)
)

使用 bigframes.ml 库

借助 BigQuery DataFrames 中的机器学习功能，您可以预处理数据，然后基于该数据训练模型。您还可以将这些操作链接在一起以创建数据流水线。

机器学习位置

bigframes.ml 支持与 BigQuery ML 相同的位置。所有 BigQuery 区域都支持 BigQuery ML 模型预测和其他机器学习函数。对模型训练的支持因区域而异。如需了解详情，请参阅 BigQuery ML 位置。

预处理数据

使用 bigframes.ml.preprocessing 模块和 bigframes.ml.compose 模块创建 Transformer（转换器），以便准备好数据以用于 Estimator（模型）。BigQuery DataFrames 提供以下转换：

  • 使用 bigframes.ml.preprocessing 模块中的 KBinsDiscretizer 类将连续数据按间隔分箱。

  • 使用 bigframes.ml.preprocessing 模块中的 LabelEncoder 类将目标标签标准化为整数值。

  • 使用 bigframes.ml.preprocessing 模块中的 MaxAbsScaler 类将每个特征按其最大绝对值缩放到 [-1, 1] 范围。

  • 使用 bigframes.ml.preprocessing 模块中的 MinMaxScaler 类，通过将每个特征缩放到 [0, 1] 范围来标准化特征。

  • 使用 bigframes.ml.preprocessing 模块中的 StandardScaler 类，通过移除平均值并缩放到单位方差来标准化特征。

  • 使用 bigframes.ml.preprocessing 模块中的 OneHotEncoder 类将分类值转换为数字格式。

  • 使用 bigframes.ml.compose 模块中的 ColumnTransformer 类将 Transformer 应用于 DataFrame 列。

训练模型

创建 Estimator 以在 BigQuery DataFrames 中训练模型。

聚类模型

使用 bigframes.ml.cluster 模块为聚类模型创建 Estimator。

  • 使用 KMeans 类创建 K-means 聚类模型。使用这些模型进行数据细分。例如，标识客户细分。K-means 是一种非监督式学习技术，因此模型训练不需要标签，也不需要为训练或评估拆分数据。

您可以使用 bigframes.ml.cluster 模块为聚类模型创建 Estimator。

以下代码示例展示如何使用 bigframes.ml.cluster KMeans 类创建用于数据细分的 k-means 聚类模型：

from bigframes.ml.cluster import KMeans
import bigframes.pandas as bpd

# Load data from BigQuery
query_or_table = "bigquery-public-data.ml_datasets.penguins"
bq_df = bpd.read_gbq(query_or_table)

# Create the KMeans model
cluster_model = KMeans(n_clusters=10)
cluster_model.fit(bq_df["culmen_length_mm"], bq_df["sex"])

# Predict using the model
result = cluster_model.predict(bq_df)
# Score the model
score = cluster_model.score(bq_df)

分解模型

使用 bigframes.ml.decomposition 模块为分解模型创建 Estimator。

  • 使用 PCA 类创建主成分分析 (PCA) 模型。使用这些模型计算主成分并使用主成分对数据执行基变换。这样便可实现降维，具体方法是将每个数据点仅投影到前几个主成分上，以获得维度较低的数据，并保留尽可能多的数据差异性。

集成学习模型

使用 bigframes.ml.ensemble 模块为集成学习模型创建 Estimator。

  • 使用 RandomForestClassifier 类创建随机森林分类器模型。使用这些模型构建多个学习方法决策树以用于分类。

  • 使用 RandomForestRegressor 类创建随机森林回归模型。使用这些模型构建多个学习方法决策树以用于回归。

  • 使用 XGBClassifier 类创建梯度提升树分类器模型。使用这些模型以累加方式构建多个学习方法决策树以用于分类。

  • 使用 XGBRegressor 类创建梯度提升树回归模型。使用这些模型以累加方式构建多个学习方法决策树以用于回归。

预测模型

使用 bigframes.ml.forecasting 模块为预测模型创建 Estimator。

导入的模型

使用 bigframes.ml.imported 模块为导入的模型创建 Estimator。

线性模型

使用 bigframes.ml.linear_model 模块为线性模型创建 Estimator。

  • 使用 LinearRegression 类创建线性回归模型。使用这些模型进行预测。例如，预测给定日期的商品销售额。

  • 使用 LogisticRegression 类创建逻辑回归模型。使用这些模型对两个或多个可能值进行分类，例如输入是 low-value、medium-value 还是 high-value。

以下代码示例展示如何使用 bigframes.ml 执行以下操作：

from bigframes.ml.linear_model import LinearRegression
import bigframes.pandas as bpd

# Load data from BigQuery
query_or_table = "bigquery-public-data.ml_datasets.penguins"
bq_df = bpd.read_gbq(query_or_table)

# Filter down to the data to the Adelie Penguin species
adelie_data = bq_df[bq_df.species == "Adelie Penguin (Pygoscelis adeliae)"]

# Drop the species column
adelie_data = adelie_data.drop(columns=["species"])

# Drop rows with nulls to get training data
training_data = adelie_data.dropna()

# Specify your feature (or input) columns and the label (or output) column:
feature_columns = training_data[
    ["island", "culmen_length_mm", "culmen_depth_mm", "flipper_length_mm", "sex"]
]
label_columns = training_data[["body_mass_g"]]

test_data = adelie_data[adelie_data.body_mass_g.isnull()]

# Create the linear model
model = LinearRegression()
model.fit(feature_columns, label_columns)

# Score the model
score = model.score(feature_columns, label_columns)

# Predict using the model
result = model.predict(test_data)

大语言模型

使用 bigframes.ml.llm 模块为 LLM 创建 Estimator。

您可以使用 bigframes.ml.llm 模块为远程大语言模型 (LLM) 创建 Estimator。
以下代码示例展示如何使用 bigframes.ml.llm GeminiTextGenerator 类创建用于生成代码的 Gemini 模型：

from bigframes.ml.llm import GeminiTextGenerator
import bigframes.pandas as bpd

# Create the Gemini LLM model
session = bpd.get_global_session()
connection = f"{PROJECT_ID}.{REGION}.{CONN_NAME}"
model = GeminiTextGenerator(session=session, connection_name=connection)

df_api = bpd.read_csv("gs://cloud-samples-data/vertex-ai/bigframe/df.csv")

# Prepare the prompts and send them to the LLM model for prediction
df_prompt_prefix = "Generate Pandas sample code for DataFrame."
df_prompt = df_prompt_prefix + df_api["API"]

# Predict using the model
df_pred = model.predict(df_prompt.to_frame(), max_output_tokens=1024)

远程模型

如需使用 BigQuery DataFrames 机器学习远程模型（bigframes.ml.remote 或 bigframes.ml.llm），您必须启用以下 API：

您还必须拥有项目的以下 IAM 角色：

  • BigQuery Connection Admin (roles/bigquery.connectionAdmin)
  • Project IAM Admin (roles/resourcemanager.projectIamAdmin)（如果使用默认 BigQuery 连接）或 Browser (roles/browser)（如果使用预配置连接）。可以通过将 bigframes.pandas.options.bigquery.skip_bq_connection_check 选项设置为 True 来避免此要求，在这种情况下，连接（默认连接或预配置连接）将按原样使用，不会进行任何存在或权限检查。如果您使用的是预配置连接并跳过连接检查，请确保在正确的位置创建连接并且其服务账号具有项目的 Vertex AI User (roles/aiplatform.user) 角色。

在 BigQuery DataFrames 中创建远程模型会创建 BigQuery 连接。默认情况下，系统会使用名为 bigframes-default-connection 的连接。 如果您愿意，可以使用预配置的 BigQuery 连接，在这种情况下会跳过连接创建。系统会向默认连接的服务账号授予 Vertex AI User (roles/aiplatform.user) IAM 角色。

创建流水线

使用 bigframes.ml.pipeline 模块创建机器学习流水线。借助流水线，您可以在设置不同的参数时组合多个要进行交叉验证的机器学习步骤。这样可简化代码，并且允许您一起部署数据预处理步骤和 Estimator。

使用流水线类创建具有最终 Estimator 的转换流水线。

后续步骤