Cloud Functions执行间隔设置及批量上传数据丢失问题求助

根因诊断 #

批量上传文件时部分行丢失，核心原因是Google Sheets的并发写入竞争。当多个Cloud Functions实例同时执行values_append操作时，Sheets API无法正确处理并发请求，导致部分写入被覆盖或静默失败。调整实例并发数和CPU配置无法解决这个本质问题。

关于设置执行间隔的问题 #

无法直接在Cloud Functions的不同执行实例之间设置1秒暂停——因为Cloud Functions是事件驱动的，每个文件上传事件会独立触发函数执行，函数本身无法控制其他实例的启动时机。强行在函数内添加time.sleep(1)反而会增加执行时长、提高成本，且无法从根本上避免竞争。

有效解决方案 #

1. 为写入操作添加重试机制 #

使用带有指数退避的重试逻辑，处理并发写入失败的情况。修改hello_gcs函数中写入Sheets的部分：

import time
from tenacity import retry, stop_after_attempt, wait_exponential

# 为写入操作添加重试装饰器（最多重试3次，间隔指数增长）
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def append_to_sheet(values):
    gs.values_append('Total', {'valueInputOption': 'RAW'}, {'values': values})

# 替换原有的写入代码
append_to_sheet(df_values)

2. 实现分布式锁控制并发写入 #

通过Cloud Firestore实现分布式锁，确保同一时间只有一个实例能写入Sheets：

from google.cloud import firestore

db = firestore.Client()
LOCK_COLLECTION = "sheet_locks"
LOCK_DOCUMENT = "write_lock"

def acquire_lock():
    try:
        # 创建锁文档，设置10秒超时（防止实例崩溃导致锁永久持有）
        db.collection(LOCK_COLLECTION).document(LOCK_DOCUMENT).set(
            {"locked": True}, merge=True, timeout=10
        )
        return True
    except firestore.exceptions.FirestoreError:
        return False

def release_lock():
    try:
        db.collection(LOCK_COLLECTION).document(LOCK_DOCUMENT).delete(timeout=10)
    except firestore.exceptions.FirestoreError:
        pass

def hello_gcs(bucket: str, filename: str):
    # ... 原有数据处理代码 ...
    
    # 尝试获取锁，最多等待5秒
    lock_acquired = False
    wait_time = 0
    while not lock_acquired and wait_time < 5:
        lock_acquired = acquire_lock()
        if not lock_acquired:
            time.sleep(1)
            wait_time +=1
    
    if lock_acquired:
        try:
            gs.values_append('Total', {'valueInputOption': 'RAW'}, {'values': df_values})
        finally:
            release_lock()
    else:
        print("无法获取锁，写入失败，请稍后重试")

3. 批量处理事件（推荐） #

将文件上传事件转发到Pub/Sub，再由一个Cloud Functions实例批量拉取事件并一次性写入Sheets，彻底避免并发竞争：

• 创建Pub/Sub主题，修改原Cloud Functions将文件信息发送到主题，而非直接写入Sheets
• 创建新的Cloud Functions订阅该主题，设置批量拉取（例如每次拉取10条消息），收集所有数据后一次性调用values_append写入Sheets

4. 临时存储中转 #

先将每行数据写入Cloud Firestore或BigQuery，再通过定时Cloud Functions将数据同步到Sheets。这种方式适合对实时性要求不高的场景，且能更好地处理大规模数据。

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