从 BQ 到云 SQL 的数据传输答案

【问题标题】：Data Transfer from BQ To CLoud SQL从 BQ 到云 SQL 的数据传输
【发布时间】：2021-12-23 08:24:00
【问题描述】：

每天将所有记录从 BigQuery 表传输到 Cloud SQL 表的最佳方法是什么（每天预计的大约记录数超过 255801312 [2.55 亿]）。我知道我们可以创建从 BQ 到 CloudSQL 的数据流管道，但是如此大量的数据将运行数小时和数小时。在谷歌云中实施的任何最佳解决方案？

【问题讨论】：

您是否尝试过从 BigQuery 以 CSV 格式导出数据，然后将它们导入 Cloud SQL？这是最有效的方法
将数据导出到 csv 并导入到 cloudsql 是一个手动过程，不确定我们是否可以按照这种方法进行每天读取 2.55 亿条记录的维护。
手动？或workflow ;)
@guillaumeblaquiere，我看到了你的帖子：medium.com/google-cloud/… EXPORT DATA OPTIONS 可能是正确的选择，但请指导我通过工作流程（一种数据流作业）运行的最佳方式是什么还可以帮助将文件从谷歌存储桶导入到 Cloudsql
工作流编排 API 调用。首先调用 BigQuery 导出存储桶中的数据，然后调用 Cloud SQL API 导入文件。

标签： google-cloud-platform google-cloud-sql bq google-cloud-data-transfer

【解决方案1】：

这是一个工作流的工作示例。您需要为您的工作流服务帐户（cloudsql 管理员、bigquery dataviewer + 作业用户、云存储管理员）提供足够的权限，并且该表必须存在于您的 Cloud SQL 实例中（我使用 MySQL 进行了测试）。

这篇文章正在烹饪，其中包含更多细节。 替换存储桶、projectid、Cloud SQL 实例名称（在我的例子中是 mysql）、查询、表名、数据库架构

main:
  steps:
    - assignStep:
        assign:
          - bucket: "TODO"
          - projectid: "TODO"
          - prefix: "workflow-import/export"
          - listResult:
              nextPageToken: ""
    - export-query:
        call: googleapis.bigquery.v2.jobs.query
        args:
          projectId: ${projectid}
          body:
            query: ${"EXPORT DATA OPTIONS( uri='gs://" + bucket + "/" + prefix + "*.csv', format='CSV', overwrite=true,header=false) AS SELECT id, email FROM `copy_dataset.name_test`"}
            useLegacySql: false
    - importfiles:
        call: import_files
        args:
          pagetoken: ${listResult.nextPageToken}
          bucket: ${bucket}
          prefix: ${prefix}
          projectid: ${projectid}
        result: listResult
    - missing-files:
        switch:
          - condition:  ${"nextPageToken" in listResult}
            next: importfiles


import_files:
  params:
    - pagetoken
    - bucket
    - prefix
    - projectid
  steps:
    - list-files:
        call: googleapis.storage.v1.objects.list
        args:
          bucket: ${bucket}
          pageToken: ${pagetoken}
          prefix: ${prefix}
        result: listResult
    - process-files:
        for:
          value: file
          in: ${listResult.items}
          steps:
            - wait-import:
                call: load_file
                args:
                  projectid: ${projectid}
                  importrequest:
                    importContext:
                      uri: ${"gs://" + bucket + "/" + file.name}
                      database: "test_schema"
                      fileType: CSV
                      csvImportOptions:
                        table: "workflowimport"
    - return-step:
        return: ${listResult}


load_file:
  params: [importrequest,projectid]
  steps:
    - callImport:
        call: http.post
        args:
          url: ${"https://sqladmin.googleapis.com/v1/projects/" + projectid + "/instances/mysql/import"}
          auth:
            type: OAuth2
          body: ${importrequest}
        result: operation
    - chekoperation:
        switch:
          - condition: ${operation.body.status != "DONE"}
            next: wait
        next: completed
    - completed:
        return: "done"
    - wait:
        call: sys.sleep
        args:
          seconds: 5
        next: getoperation
    - getoperation:
        call: http.get
        args:
          url: ${operation.body.selfLink}
          auth:
            type: OAuth2
        result: operation
        next: chekoperation

更多详情my medium article

【讨论】：