如何将Google Sheets文件夹中最新表格数据同步至Redshift？

Absolutely, you can tweak your existing workflow to automatically pull the newest Google Sheet from a specified Drive folder and load it into your Redshift database. Let’s walk through this with actionable code that builds on your current implementation.

Step 1: Prep Google Drive API Access #

First, you’ll need to enable the Google Drive API in your Google Cloud Project (you probably already have the Sheets API enabled). Also, make sure your service account (or OAuth credentials) has edit access to the target Drive folder—share the folder with the service account’s email address found in your JSON key file.

Step 2: Fetch the Latest Sheet from the Folder #

Add this function to retrieve the most recently created (or modified) Google Sheet in your target folder:

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

def get_latest_sheet_in_folder(folder_id, credentials_path):
    # Load credentials (reuse your existing auth setup)
    creds = Credentials.from_service_account_file(
        credentials_path,
        scopes=['https://www.googleapis.com/auth/drive.readonly', 'https://www.googleapis.com/auth/spreadsheets.readonly']
    )
    
    # Initialize Drive API client
    drive_service = build('drive', 'v3', credentials=creds)
    
    # Query to get all Google Sheets in the target folder, sorted by creation time (newest first)
    query = f"'{folder_id}' in parents and mimeType='application/vnd.google-apps.spreadsheet'"
    results = drive_service.files().list(
        q=query,
        orderBy='createdTime desc',  # Use 'modifiedTime desc' if you want last edited instead
        fields='files(id, name, createdTime)'
    ).execute()
    
    items = results.get('files', [])
    if not items:
        raise ValueError("No Google Sheets found in the specified folder.")
    
    # Return the latest sheet's ID and name
    latest_sheet = items[0]
    print(f"Found latest sheet: {latest_sheet['name']} (ID: {latest_sheet['id']}, created: {latest_sheet['createdTime']})")
    return latest_sheet['id']

Step 3: Read Data from the Latest Sheet #

Reuse your existing Sheets API logic, but replace the hardcoded spreadsheet ID with the one returned from the function above. Here’s a quick example using pandas (adjust to match your current read method):

import pandas as pd
from googleapiclient.discovery import build

def read_sheet_data(spreadsheet_id, creds):
    sheets_service = build('sheets', 'v4', credentials=creds)
    
    # Read data from the first worksheet (adjust range if needed, e.g., 'Sheet1!A:Z')
    sheet = sheets_service.spreadsheets()
    result = sheet.values().get(spreadsheetId=spreadsheet_id, range='A:Z').execute()
    values = result.get('values', [])
    
    if not values:
        raise ValueError("No data found in the sheet.")
    
    # Convert to DataFrame (use first row as headers)
    df = pd.DataFrame(values[1:], columns=values[0])
    return df

Step 4: Write Data to Redshift #

Use psycopg2 or SQLAlchemy to load the DataFrame into Redshift. Below is an example with SQLAlchemy (simpler for DataFrame operations):

First, install required packages if you haven’t already:

pip install sqlalchemy psycopg2-binary pandas

Then the code:

from sqlalchemy import create_engine

def write_to_redshift(df, redshift_conn_str, table_name, if_exists='replace'):
    # Create Redshift engine
    engine = create_engine(redshift_conn_str)
    
    # Write DataFrame to Redshift
    df.to_sql(
        name=table_name,
        con=engine,
        if_exists=if_exists,  # Options: 'replace', 'append', 'fail'
        index=False,
        method='multi'  # Faster for bulk inserts
    )
    print(f"Successfully wrote {len(df)} rows to Redshift table {table_name}")

Full Workflow Example #

Put it all together in a main script:

def main():
    # Configuration
    CREDENTIALS_PATH = 'path/to/your/service-account-key.json'
    DRIVE_FOLDER_ID = 'your-target-folder-id'  # Find this in the Drive folder URL
    REDSHIFT_CONN_STR = 'postgresql+psycopg2://username:password@redshift-cluster-url:5439/database-name'
    REDSHIFT_TABLE_NAME = 'your_target_table'
    
    try:
        # Get latest sheet ID
        creds = Credentials.from_service_account_file(
            CREDENTIALS_PATH,
            scopes=['https://www.googleapis.com/auth/drive.readonly', 'https://www.googleapis.com/auth/spreadsheets.readonly']
        )
        latest_sheet_id = get_latest_sheet_in_folder(DRIVE_FOLDER_ID, CREDENTIALS_PATH)
        
        # Read sheet data
        df = read_sheet_data(latest_sheet_id, creds)
        
        # Clean data if needed (e.g., convert data types, handle missing values)
        # df['date_column'] = pd.to_datetime(df['date_column'])
        # df['numeric_column'] = pd.to_numeric(df['numeric_column'], errors='coerce')
        
        # Write to Redshift
        write_to_redshift(df, REDSHIFT_CONN_STR, REDSHIFT_TABLE_NAME, if_exists='replace')
        
    except Exception as e:
        print(f"Error during workflow: {str(e)}")

if __name__ == "__main__":
    main()

Key Notes & Troubleshooting #

• Permissions: Ensure your service account has access to the Drive folder (share it with the service account email from your JSON key).
• Time Sorting: Use orderBy='modifiedTime desc' instead of createdTime if you want the last-edited sheet instead of the newly created one.
• Data Type Matching: Make sure your Redshift table schema matches the data types in your Google Sheet (e.g., dates, numbers, text lengths). Add data cleaning steps in the workflow to align types.
• Large Datasets: For very large sheets, consider using Redshift’s COPY command (export Sheet data to CSV in Drive, upload to S3, then COPY to Redshift) for better performance.

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