第八篇: 通过使用Google BigQuery进行数据批量和自动化处理

使用Python进行Google BigQuery数据批量和自动化处理

在大数据分析的日常工作中，定期更新、查询和处理数据是一项必不可少的任务。Google BigQuery结合Python脚本，可大幅简化这一过程。本文将介绍如何通过Python自动查询和更新BigQuery中的降水量数据，适用于需要定期获取最新信息的数据分析场景。

我们将基于Google的公共数据集中的天气降水量数据集来进行实际演示，并实现一个Python脚本自动化数据更新的示例。

数据集选择及准备

Google提供了众多免费的公共数据集，其中的“NOAA Global Surface Summary of the Day Weather Data”包含各地的历史天气信息，包括降水量。这个数据集适合我们演示批量查询和更新流程。

连接BigQuery并检查表结构：
- 首先，确保Python已安装Google Cloud BigQuery库。使用以下命令安装BigQuery客户端库：
```
pip install google-cloud-bigquery
```
- 在脚本中导入bigquery库并设置连接。以下是基本的连接代码：
```
from google.cloud import bigquery

# 初始化BigQuery客户端
client = bigquery.Client()
```
查询降水量数据

假设我们希望获取某地最近一个月的每日降水量，可以使用SQL查询。例如，假设我们查询的是数据集中“日降水量”和“观测日期”等字段。

SELECT 
    date,
    stn,
    prcp
FROM 
    `bigquery-public-data.noaa_gsod.gsod2024` 
WHERE 
    stn = '388270'  -- 例如：某观测站
    AND prcp <> 0
    AND date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
ORDER BY 
    date

查询结果示例：

date	stn	prcp
2024-10-15	388270	99.99
2024-10-16	388270	0.12
2024-10-22	388270	99.99
2024-10-23	388270	0.24
2024-10-25	388270	0.16
2024-10-26	388270	0.47
2024-10-27	388270	99.99

在Python中自动化运行查询

现在我们可以在Python中设置一个自动化脚本，每月初执行一次上面的查询，以获取过去一个月的降水量数据。下面的Python代码演示了如何实现自动化查询，并将查询结果存储为CSV文件。

from google.cloud import bigquery
import pandas as pd
import datetime

# 初始化BigQuery客户端
client = bigquery.Client()

# 编写SQL查询
query = """
SELECT 
 date,
 stn,
 prcp
 FROM 
     `bigquery-public-data.noaa_gsod.gsod2024` 
 WHERE 
     stn = '388270'  -- 例如：纽约某观测站
     AND prcp <> 0
     AND date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
 ORDER BY 
     date

"""

# 执行查询并将结果存储为DataFrame
query_job = client.query(query)  # Make an API request
results = query_job.result().to_dataframe()  # 将结果转换为DataFrame

# 保存结果到CSV
results.to_csv("monthly_precipitation_data.csv", index=False)
print("Data saved to monthly_precipitation_data.csv")

以上脚本会将过去30天的降水量数据保存为CSV文件，每次执行时可自动更新。

为了实现定期更新，可以将该Python脚本配置为每月自动运行一次，获取最新数据。这里我们推荐使用 cron（Linux系统）或 Task Scheduler（Windows系统），设置定期执行脚本的任务。

在Linux上，可以添加以下cron任务，设置脚本在每月1日的凌晨运行一次：

bash
复制代码
0 0 1 * * /usr/bin/python3 /path/to/your_script.py

通过BigQuery定时任务自动化更新

Google BigQuery提供的定时任务可以让我们在云端定期运行查询并更新数据集。此方法适用于企业应用中需要更高频次的数据更新。以下是设置BigQuery定时查询的步骤：

创建定时查询：
- 登录Google Cloud Console，转到BigQuery页面。
- 在查询编辑器中输入SQL语句。
- 点击“保存查询”，并选择“设置定时查询”。
配置频率和目标：
- 设置定时任务的运行频率（例如，每天运行）。
- 指定查询结果的存储位置，例如将结果保存到新的BigQuery表或导出到Google Cloud Storage。