Vertex AI调度任务无法推送数据至BigQuery，手动运行正常

1. 检查调度任务的服务账号权限 #

• 手动运行代码时使用的是个人账号，通常拥有较全权限；但Vertex AI调度任务默认使用专属服务账号，可能缺少BigQuery写入权限。
• 找到任务使用的服务账号（默认是service-{你的项目编号}@gcp-sa-vertexai.iam.gserviceaccount.com），在IAM中为其添加BigQuery Data Editor和BigQuery Job User角色，确保它能创建BigQuery任务并写入数据。

2. 确认运行环境的依赖是否完整 #

• 手动运行时你的环境已安装requests、pandas等依赖，但Vertex AI调度任务的环境可能默认未包含这些包。
• 解决方式：
  • 如果使用预构建镜像，需在代码目录中添加requirements.txt，内容包含：

    requests>=2.31.0
    pandas>=2.0.0
    google-cloud-bigquery>=3.11.4
    

  • 如果是Notebook生成的调度任务，需先在Notebook中运行!pip install requests pandas，再保存环境后创建调度任务。

3. 排查任务触发与超时配置 #

• 检查调度任务的时区设置：确认触发时间的时区与预期一致（默认是UTC，若你按本地时区设置，可能导致任务还未到执行时间）。
• 延长任务超时时间：默认超时可能较短（如1小时），若数据采集或写入耗时较长，任务会被静默终止且无明显日志。在创建调度任务时，将超时时间调整为合适长度（如2小时）。

4. 查看Cloud Logging的隐藏日志 #

• 任务状态显示“已调度”但无执行记录，可能是任务执行后出现了未被捕获的错误，需到Cloud Logging中排查：
  • 进入Cloud Logging，过滤条件选择resource.type="aiplatform_job"，并输入任务ID进行搜索，查看是否有网络异常、权限错误等隐性日志。

5. 优化代码的异常捕获逻辑 #

• 在代码中添加异常捕获，确保错误能被记录到日志中，方便排查：
  修改后的代码片段：

  # Scrape the data and load it into BigQuery
  for url in urls:
      try:
          r = requests.get(url, timeout=15)
          r.raise_for_status()  # 触发HTTP状态码错误（如403、500）
          data = r.json()
      except Exception as e:
          print(f"采集URL {url} 失败: {str(e)}")
          raise  # 抛出错误，让任务记录异常日志
  
      rows = []
      for item in data["products"]:
          title = item["title"]
          created = item["created_at"]
          for variant in item["variants"]:
              try:
                  price = float(variant["price"])
                  sku = str(variant["sku"])
                  available = bool(variant["available"])
                  row = {
                      'title': str(title),
                      'created': str(created),
                      'price': price,
                      'sku': sku,
                      'available': available,
                      'retrieval_date': date.today().isoformat()
                  }
                  rows.append(row)
              except KeyError as e:
                  print(f"解析商品变体失败，缺失字段: {str(e)}")
                  continue  # 跳过异常数据，继续处理其他变体
  
      # 写入BigQuery的代码也添加异常捕获
      try:
          job_config = bigquery.LoadJobConfig(schema=schema, write_disposition=bigquery.WriteDisposition.WRITE_APPEND)
          job = client.load_table_from_json(rows, table_ref, job_config=job_config)
          job.result()  # 等待任务完成，捕获写入错误
          print(f"成功写入 {len(rows)} 条数据到BigQuery")
      except Exception as e:
          print(f"写入BigQuery失败: {str(e)}")
          raise
  

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