Python实现读取Excel校验BigQuery表存在性并回写结果

我正在学习Python，现有一份记录多个BigQuery表信息的Excel文件，希望通过Python读取该文件，校验这些表是否存在于BigQuery中，并将结果回写到Excel。目前无法实现多行列循环，请求技术指导。

原代码 #

import pandas as pd
from google.cloud import bigquery
from google.oauth2 import service_account
import openpyxl
from openpyxl.utils.dataframe import dataframe_to_rows
from google.cloud.exceptions import NotFound
import numpy


credentials = service_account.Credentials.from_service_account_file('file path.json')

project_id = 'creating-data-413804'
client = bigquery.Client(credentials= credentials,project=project_id)

#Reading the excel file 

data = pd.read_excel('file path', usecols=('Dataset','Source Schema', 'Source Table Name', 'Target   BQ Project ID', 'table exist'))

for col in data.iterrows():

    project_nm = data['Source Schema'][col]
    dataset_nm =data['Dataset'][col]
    table_nm = data['Source Table Name'][col]
    client = bigquery.Client(credentials= credentials,project=project_id)
    dataset = client.dataset(dataset_nm)
    table_ref = dataset.table(table_nm)
    def if_tbl_exists(client, table_ref):
        #from google.cloud.exceptions import NotFound
        try:
            client.get_table(table_ref)
            return True
        except NotFound:
            return False
print(col)
print(if_tbl_exists(client, table_ref))

book = openpyxl.load_workbook('file path.xlsx')  

sheet = book['Sheet1']
if if_tbl_exists(client, table_ref) == True:
    sheet.cell(row=3, column=5).value = 'table exist'
else:
    sheet.cell(row=3, column=5).value = 'table doesnt exist'

book.save('file path.xlsx')

问题分析 #

• iterrows() 返回的是**(行索引, 行数据)**的元组，原代码直接用col接收，导致取数逻辑错误
• 校验函数if_tbl_exists定义在循环内部，会重复创建，不符合代码规范
• 硬编码写死行号row=3，无法处理多行数据
• 重复创建bigquery.Client实例，浪费资源
• 列名匹配错误：原代码用Source Schema作为项目名，但Excel中存在Target BQ Project ID列，应该用该列指定目标校验项目

修正后的代码 #

import pandas as pd
from google.cloud import bigquery
from google.oauth2 import service_account
import openpyxl
from google.cloud.exceptions import NotFound

# 初始化BigQuery客户端（仅创建一次）
credentials = service_account.Credentials.from_service_account_file('file path.json')
client = bigquery.Client(credentials=credentials)

# 定义表存在校验函数（放在循环外部）
def if_tbl_exists(client, project_id, dataset_id, table_id):
    try:
        table_ref = client.dataset(dataset_id, project=project_id).table(table_id)
        client.get_table(table_ref)
        return 'table exist'
    except NotFound:
        return 'table doesnt exist'

# 读取Excel数据
excel_path = 'file path.xlsx'
data = pd.read_excel(
    excel_path,
    usecols=['Dataset', 'Source Table Name', 'Target BQ Project ID', 'table exist']
)

# 循环处理每一行数据，更新校验结果
for idx, row in data.iterrows():
    # 获取当前行的项目、数据集、表名
    project_nm = row['Target BQ Project ID']
    dataset_nm = row['Dataset']
    table_nm = row['Source Table Name']
    
    # 执行校验并更新结果列
    data.loc[idx, 'table exist'] = if_tbl_exists(client, project_nm, dataset_nm, table_nm)

# 将结果写回Excel
book = openpyxl.load_workbook(excel_path)
sheet = book['Sheet1']

# 遍历DataFrame行，写入对应Excel单元格（Excel行号从1开始，表头占1行，数据从第2行开始）
for idx, row in data.iterrows():
    sheet.cell(row=idx+2, column=5).value = row['table exist']

# 保存文件
book.save(excel_path)

关键改进点 #

• 将校验函数移至循环外，避免重复定义
• 正确使用iterrows()遍历每行数据，通过row[列名]获取对应值
• 动态计算Excel行号（idx+2），适配多行数据处理
• 仅创建一次BigQuery客户端，提升运行效率
• 修正项目ID的列名匹配逻辑，确保校验目标项目下的表

内容的提问来源于stack exchange，提问作者Satya Anvesh