实现PowerBI中CSV文件每日自动上传BigQuery的Python库推荐

• pandas：用来读取、清洗PowerBI导出的CSV文件，处理空值、列类型转换等数据预处理工作，是数据处理的基础工具，常用方法比如pd.read_csv()读取文件，df.dropna()清理空值。
• google-cloud-bigquery：Google官方的BigQuery Python客户端，提供直接操作BigQuery的API，支持创建表、写入数据、查询等全功能；配套的pandas-gbq是pandas和BigQuery的桥接库，可以直接用df.to_gbq()把DataFrame写入BigQuery，更简洁高效。
• schedule：轻量的定时任务库，无需依赖系统定时工具（如cron、Windows任务计划），在脚本内就能设置每日固定时间执行上传逻辑，比如schedule.every().day.at("09:00").do(upload_task)。
• python-dotenv：管理敏感配置（如BigQuery服务账号密钥路径、项目ID、CSV文件路径），将这些信息存在.env文件中，脚本通过load_dotenv()读取，避免硬编码泄露信息。
• logging（可选但推荐）：记录脚本运行日志，方便排查上传失败、数据异常等问题，比如记录上传时间、数据行数、错误信息。

1. 配置BigQuery权限
   • 创建BigQuery服务账号，下载JSON格式的密钥文件，给账号分配「BigQuery数据编辑者」和「BigQuery作业用户」权限，确保能读写目标数据集。
2. 安装依赖库
   执行命令：

   pip install pandas google-cloud-bigquery pandas-gbq schedule python-dotenv
   

3. 编写核心上传脚本
   示例框架：

   import os
   import pandas as pd
   import pandas_gbq
   import schedule
   import time
   from dotenv import load_dotenv
   
   # 加载配置
   load_dotenv()
   PROJECT_ID = os.getenv("BQ_PROJECT_ID")
   DATASET_ID = os.getenv("BQ_DATASET_ID")
   TABLE_ID = os.getenv("BQ_TABLE_ID")
   CSV_PATH = os.getenv("CSV_PATH")
   KEY_PATH = os.getenv("BQ_KEY_PATH")
   
   def upload_to_bigquery():
       try:
           # 读取CSV
           df = pd.read_csv(CSV_PATH)
           # 数据预处理（根据实际需求调整）
           df = df.dropna(subset=["关键列"])
           df["日期列"] = pd.to_datetime(df["日期列"])
           
           # 写入BigQuery
           pandas_gbq.to_gbq(
               df,
               destination_table=f"{DATASET_ID}.{TABLE_ID}",
               project_id=PROJECT_ID,
               credentials=KEY_PATH,
               if_exists="append"  # 每日上传一般用追加，如需覆盖改为"replace"
           )
           print(f"上传成功：{len(df)}条数据")
       except Exception as e:
           print(f"上传失败：{str(e)}")
   
   # 设置每日定时任务
   schedule.every().day.at("08:30").do(upload_to_bigquery)
   
   # 保持脚本运行
   while True:
       schedule.run_pending()
       time.sleep(60)
   

4. 测试与部署
   • 手动运行脚本，检查BigQuery中是否成功写入数据，验证数据格式、行数是否正确。
   • 部署方式：可以将脚本设置为系统后台进程（如Linux的nohup、Windows的任务计划程序），确保每日自动执行。

• 确保PowerBI导出CSV的时间早于脚本执行时间，避免读取未生成的空文件。
• 提前对齐BigQuery表结构与CSV列的名称、数据类型，避免因类型不匹配导致上传失败。
• 添加异常捕获和日志记录，方便定位问题（比如网络波动、权限过期等）。

内容的提问来源于stack exchange，提问作者Andres Peña