BigQuery Data Transfer Service：CLI/Python客户端如何实现传输后删源文件？

BigQuery Data Transfer Service的CLI/Python客户端确实没有直接暴露“传输完成后删除源文件”的参数，针对按需任务场景，你可以通过以下两种方式实现需求：

方法一：Python脚本轮询任务状态+手动删除源文件 #

适合需要按需触发传输并同步执行删除的场景，直接在启动传输的脚本里等待任务完成后执行删除操作：

步骤说明 #

1. 初始化BigQuery Data Transfer和GCS客户端
2. 创建并启动按需传输任务，获取任务运行ID
3. 轮询任务状态，直到任务成功完成
4. 列出GCS目标路径下的所有文件并批量删除

代码示例 #

from google.cloud import bigquery_datatransfer_v1
from google.cloud import storage
import time

# 替换为你的项目信息
PROJECT_ID = "your-project-id"
LOCATION = "us-central1"
TRANSFER_CONFIG_ID = "your-transfer-config-id"  # 提前创建好的传输配置ID
BUCKET_NAME = "your-bucket-name"
SOURCE_PATH_PREFIX = "your-bucket-path/"  # 对应gs://bucket/your-bucket-path/*的前缀

# 初始化客户端
transfer_client = bigquery_datatransfer_v1.DataTransferServiceClient()
storage_client = storage.Client()

# 构建传输配置路径
transfer_config_name = transfer_client.transfer_config_path(PROJECT_ID, LOCATION, TRANSFER_CONFIG_ID)

# 启动按需传输任务
start_request = bigquery_datatransfer_v1.StartManualTransferRunsRequest(
    parent=transfer_config_name,
    requested_run_time={"seconds": int(time.time())}
)
start_response = transfer_client.start_manual_transfer_runs(start_request)
transfer_run_name = start_response.runs[0].name

# 轮询任务状态
print(f"等待传输任务完成，任务ID: {transfer_run_name}")
while True:
    transfer_run = transfer_client.get_transfer_run(name=transfer_run_name)
    if transfer_run.state == bigquery_datatransfer_v1.TransferRun.State.SUCCEEDED:
        print("传输任务成功完成")
        break
    elif transfer_run.state in [
        bigquery_datatransfer_v1.TransferRun.State.FAILED,
        bigquery_datatransfer_v1.TransferRun.State.CANCELLED
    ]:
        print(f"传输任务失败/取消，状态: {transfer_run.state.name}")
        exit(1)
    time.sleep(30)  # 每30秒检查一次状态

# 批量删除GCS源文件
bucket = storage_client.bucket(BUCKET_NAME)
blobs = list(bucket.list_blobs(prefix=SOURCE_PATH_PREFIX))
if blobs:
    bucket.delete_blobs(blobs)
    print(f"成功删除{len(blobs)}个源文件")
else:
    print("未找到需要删除的源文件")

注意事项 #

• 确保运行脚本的账号拥有bigquery.datatransfers.get（查看传输任务状态）和storage.objects.delete（删除GCS文件）的权限
• SOURCE_PATH_PREFIX需要准确匹配源URI的路径部分，比如源URI是gs://my-bucket/data/*，则前缀设为data/
• 可根据实际情况调整轮询间隔，避免过于频繁的API调用

方法二：Cloud Function + Pub/Sub监听传输完成事件 #

适合全自动化场景，无需手动运行脚本，通过事件触发自动删除：

步骤说明 #

1. 创建Pub/Sub主题，用于接收BigQuery Data Transfer的状态事件
2. 在传输配置中开启事件通知，指定上述Pub/Sub主题
3. 创建Cloud Function，订阅该Pub/Sub主题，触发时删除源文件

Cloud Function代码示例（Python） #

import base64
import json
from google.cloud import storage

def delete_gcs_files(event, context):
    # 解析Pub/Sub事件内容
    pubsub_data = base64.b64decode(event["data"]).decode("utf-8")
    event_payload = json.loads(pubsub_data)
    
    # 从事件中提取传输配置的源URI（需根据实际事件结构调整）
    transfer_params = event_payload.get("transfer_config", {}).get("params", {})
    source_uris = transfer_params.get("source_uris", [])
    if not source_uris:
        print("未找到源URI信息")
        return
    
    source_uri = source_uris[0]
    if not source_uri.startswith("gs://"):
        print(f"无效的源URI: {source_uri}")
        return
    
    # 拆分Bucket名称和文件前缀
    uri_parts = source_uri.split("/")
    bucket_name = uri_parts[2]
    # 处理gs://bucket/path/*的格式，提取前缀为path/
    file_prefix = "/".join(uri_parts[3:-1]) + "/" if len(uri_parts) > 4 else ""
    
    # 批量删除文件
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blobs = list(bucket.list_blobs(prefix=file_prefix))
    
    if blobs:
        bucket.delete_blobs(blobs)
        print(f"成功删除{len(blobs)}个文件，路径: {source_uri}")
    else:
        print(f"路径{source_uri}下无文件可删除")

注意事项 #

• 确保Cloud Function的服务账号拥有storage.objects.delete权限和Pub/Sub订阅权限
• 事件结构可能因传输类型（GCS/BigQuery等）略有不同，需根据实际事件日志调整解析逻辑
• 可在Cloud Function中添加错误处理，避免因事件解析失败导致函数报错

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