bquest 0.5.1

概述

• 将BQuest与您喜欢的测试框架（例如pytest）结合使用。

• 从JSON或pandas DataFrame创建临时测试表。

• 在您的测试表上运行BQ配置和普通SQL查询，并检查结果。

安装

通过PyPi（标准）

pip install bquest

通过Github（最新版本）

pip install git+https://github.com/ottogroup/bquest

BQuest还需要一个专用的BigQuery数据集来存储测试表，例如。

resource "google_bigquery_dataset" "bquest" {
  dataset_id    = "bquest"
  friendly_name = "bquest"
  description   = "Source tables for bquest tests"
  location      = "EU"
  default_table_expiration_ms = 3600000
}

我们建议为bquest数据集中的表设置一个过期时间，以确保在测试执行期间删除这些测试表。

示例

给定一个pandas DataFrame

以及其表定义

from bquest.tables import BQTableDefinitionBuilder

table_def_builder = BQTableDefinitionBuilder(GOOGLE_PROJECT_ID, dataset="bquest", location="EU")
table_definition = table_def_builder.from_df("abc.feed_latest", df)

您可以使用配置文件./abc/config.py

{
    "query": """
        SELECT
            foo,
            PARSE_DATE('%Y%m%d', prediction_date)
        FROM
            `{source_table}`
        WHERE
            weight > {THRESHOLD}
    """,
    "start_date": "prediction_date",
    "end_date": "prediction_date",
    "source_tables": {"source_table": "abc.feed_latest"},
    "feature_table_name": "abc.myid",
}

和运行器

from bquest.runner import BQConfigFileRunner, BQConfigRunner

runner = BQConfigFileRunner(
    BQConfigRunner(bq_client, bq_executor_func),
    "config/bq_config",
)

result_df = runner.run_config(
    "20190301",
    "20190308",
    [table_definition],
    "abc/config.py",
    templating_vars={"THRESHOLD": "30"},
)

来断言结果表

assert result_df.shape == (1, 2)
assert result_df.iloc[0]["foo"] == "my"

测试

对于实际的测试，bquest依赖于一个可访问的大数据查询项目，该项目可以通过gcloud客户端配置。相应的GOOGLE_PROJECT_ID从该项目中提取，并用于pandas-gbq，将临时表写入在测试之前需要预先配置的bquest数据集。

对于Github CI，我们在测试项目中配置了一个身份提供者，该提供者只允许该存储库的核心成员访问测试项目的资源。

重要链接

• 完整文档：https://ottogroup.github.io/bquest/