从 BigQuery 数据创建实时仪表盘

Google BigQuery 是一种基于云的服务，用于处理超大型数据集。它是一种无服务器且高度可扩展的数据仓库解决方案，使用户能够使用类似 SQL 的查询分析数据。

在本教程中，我们将向您展示如何在 Python 中查询 BigQuery 数据集，并使用 gradio 在实时更新的仪表盘中显示数据。仪表盘将如下所示

我们将在本指南中介绍以下步骤

1. 设置你的 BigQuery 凭据
2. 使用 BigQuery 客户端
3. 构建实时仪表盘（仅需 7 行 Python 代码）

我们将使用 纽约时报的 COVID 数据集，该数据集在 BigQuery 上作为公共数据集提供。该数据集名为 covid19_nyt.us_counties，包含美国各县 COVID 确诊病例和死亡人数的最新信息。

先决条件：本指南使用 Gradio Blocks，因此请确保您熟悉 Blocks 类。

设置你的 BigQuery 凭据

要将 Gradio 与 BigQuery 结合使用，您需要获取您的 BigQuery 凭据，并将其与 BigQuery Python 客户端 一起使用。如果您已经拥有 BigQuery 凭据（作为 .json 文件），则可以跳过本节。否则，您可以在几分钟内免费完成此操作。

1. 首先，登录您的 Google Cloud 帐户并转到 Google Cloud Console (https://console.cloud.google.com/)

2. 在 Cloud Console 中，单击左上角的汉堡菜单，然后从菜单中选择“API 和服务”。如果您没有现有项目，则需要创建一个。

3. 然后，单击“+ 启用 API 和服务”按钮，您可以通过该按钮为您的项目启用特定服务。搜索“BigQuery API”，单击它，然后单击“启用”按钮。如果您看到“管理”按钮，则表示 BigQuery 已启用，您已完成设置。

4. 在“API 和服务”菜单中，单击“凭据”选项卡，然后单击“创建凭据”按钮。

5. 在“创建凭据”对话框中，选择“服务帐户密钥”作为要创建的凭据类型，并为其命名。还要通过为服务帐户授予角色（例如“BigQuery 用户”）来授予其权限，这将允许您运行查询。

6. 选择服务帐户后，选择“JSON”密钥类型，然后单击“创建”按钮。这将把包含您的凭据的 JSON 密钥文件下载到您的计算机。它看起来像这样

{
	"type": "service_account",
	"project_id": "your project",
	"private_key_id": "your private key id",
	"private_key": "private key",
	"client_email": "email",
	"client_id": "client id",
	"auth_uri": "https://#/o/oauth2/auth",
	"token_uri": "https://#/o/oauth2/token",
	"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
	"client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/email_id"
}

使用 BigQuery 客户端

获得凭据后，您需要使用 BigQuery Python 客户端，通过您的凭据进行身份验证。为此，您需要在终端中运行以下命令来安装 BigQuery Python 客户端

pip install google-cloud-bigquery[pandas]

您会注意到我们已经安装了 pandas 附加组件，这将有助于将 BigQuery 数据集处理为 pandas 数据帧。安装客户端后，您可以通过运行以下代码使用您的凭据进行身份验证

from google.cloud import bigquery

client = bigquery.Client.from_service_account_json("path/to/key.json")

通过凭据验证，您现在可以使用 BigQuery Python 客户端与您的 BigQuery 数据集进行交互。

这是一个函数的示例，该函数查询 BigQuery 中的 covid19_nyt.us_counties 数据集，以显示截至当前日期确诊病例最多的前 20 个县

import numpy as np

QUERY = (
    'SELECT * FROM `bigquery-public-data.covid19_nyt.us_counties` '
    'ORDER BY date DESC,confirmed_cases DESC '
    'LIMIT 20')

def run_query():
    query_job = client.query(QUERY)
    query_result = query_job.result()
    df = query_result.to_dataframe()
    # Select a subset of columns
    df = df[["confirmed_cases", "deaths", "county", "state_name"]]
    # Convert numeric columns to standard numpy types
    df = df.astype({"deaths": np.int64, "confirmed_cases": np.int64})
    return df

构建实时仪表盘

一旦您有了查询数据的函数，就可以使用 Gradio 库中的 gr.DataFrame 组件以表格格式显示结果。这是一种检查数据并确保数据查询正确的有用方法。

以下是如何使用 gr.DataFrame 组件显示结果的示例。通过将 run_query 函数传递给 gr.DataFrame，我们指示 Gradio 在页面加载后立即运行该函数并显示结果。此外，您还可以传入关键字 every 来告诉仪表盘每小时刷新一次（60*60 秒）。

import gradio as gr

with gr.Blocks() as demo:
    gr.DataFrame(run_query, every=gr.Timer(60*60))

demo.launch()

也许你想在我们的仪表盘中添加可视化效果。您可以使用 gr.ScatterPlot() 组件在散点图中可视化数据。这使您能够看到数据集中不同变量（如病例数和病例死亡数）之间的关系，并且对于探索数据和获得见解非常有用。同样，我们可以通过传入 every 参数来实时执行此操作。

这是一个完整的示例，展示了如何使用 gr.ScatterPlot 进行可视化，以及如何使用 gr.DataFrame 显示数据

import gradio as gr

with gr.Blocks() as demo:
    gr.Markdown("# 💉 Covid Dashboard (Updated Hourly)")
    with gr.Row():
        gr.DataFrame(run_query, every=gr.Timer(60*60))
        gr.ScatterPlot(run_query, every=gr.Timer(60*60), x="confirmed_cases",
                        y="deaths", tooltip="county", width=500, height=500)

demo.queue().launch()  # Run the demo with queuing enabled