如何每日自动将Databricks数据同步至Google Sheets？代码问题求助

需求可行性 #

完全可行。你可以通过以下步骤实现目标：

• 在Databricks中编写数据提取逻辑，将数据导出为CSV/Excel等通用格式
• 借助Google Drive API将文件上传至指定目录，或直接写入Google Sheets
• 通过Databricks Jobs配置每日定时任务，自动执行整个数据同步流程

代码问题分析与修复 #

你的代码存在多个语法、逻辑错误，其中凭证文件读取失败是核心问题之一，以下是修正后的完整代码及问题说明：

关键问题点 #

1. DBFS路径无法直接读取：Databricks的dbfs:/路径不能直接传给本地文件读取函数，需转换为/dbfs/开头的本地可访问路径
2. 凭证加载语法错误：from_service_account_file的调用括号位置错误，导致未正确传入凭证路径
3. 类方法归属错误：share_file方法缩进错误，未被包含在Sender类中
4. API调用逻辑错误：send_file中execute()后直接调用print，会导致返回值变为None，后续调用get方法会报错
5. 初始化方法重复调用：__init__中两次调用set_drive_service，且第二次传参但方法未定义参数

修正后的代码 #

from google.oauth2 import service_account
from googleapiclient.discovery import build

class Sender():
    def __init__(self, cred_path):
        self.drive_service = None
        self.set_drive_service(cred_path)

    def set_drive_service(self, cred_path):
        # 转换DBFS路径为本地可访问路径
        if cred_path.startswith("dbfs:/"):
            local_cred_path = cred_path.replace("dbfs:/", "/dbfs/")
        else:
            local_cred_path = cred_path
        
        # 正确加载服务账号凭证，并添加Drive API权限范围
        credentials = service_account.Credentials.from_service_account_file(local_cred_path)
        scoped_credentials = credentials.with_scopes([
            "https://www.googleapis.com/auth/drive.file"
        ])
        self.drive_service = build('drive', 'v3', credentials=scoped_credentials)
    
    def send_file(self, drive_folder_id, filepath):
        # 处理数据文件的DBFS路径
        if filepath.startswith("dbfs:/"):
            local_filepath = filepath.replace("dbfs:/", "/dbfs/")
        else:
            local_filepath = filepath
            
        filename = local_filepath.split("/")[-1]

        file_metadata = {
            "name": filename,
            "parents": [drive_folder_id]
        }

        # 先执行API上传，再打印结果
        file_response = self.drive_service.files().create(
            body=file_metadata, 
            media_body=local_filepath, 
            fields="id"
        ).execute()
        print(f"上传文件ID: {file_response.get('id')}")
        return file_response.get("id")
    
    def share_file(self, id_file, email):
        permission = {
            'type': 'user',
            'role': 'reader',
            'emailAddress': email,
            'sendNotificationEmail': False
        }    

        self.drive_service.permissions().create(
            fileId=id_file, 
            body=permission, 
            fields='id'
        ).execute()

# 使用示例（替换为你的实际参数）
if __name__ == "__main__":
    sender = Sender("dbfs:/FileStore/credentials/credentials.json")
    uploaded_file_id = sender.send_file("你的Google Drive目标文件夹ID", "dbfs:/FileStore/extracted_data.csv")
    sender.share_file(uploaded_file_id, "需要查看数据的用户邮箱")

额外注意事项 #

• 确保Google服务账号凭证已获得Google Drive的访问权限，且目标Drive文件夹已共享给该服务账号的邮箱
• 在Databricks Jobs中运行时，需提前安装依赖包：pip install google-auth google-api-python-client
• 若要直接写入Google Sheets而非上传文件，可使用gspread库结合服务账号凭证实现更高效的数据同步

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