GCS から、CLoud SQL への連携

全体像

この文書では、Cloud Storage （GCS）にアップロードした CSV ファイルを Cloud SQL にインポートする方法について記述します。とくに、Eventarc trigger で GCS への CSV ファイルのアップロードを検出し、ワークフローを実行する方法を解説します。

事前準備

GCS の準備

Eventarc trigger でイベントを補足する対象となるバケットを準備します。手順がわからない方は、Cloud Storage などを参照してください。

ワークフローの作成

Workflows からワークフローを作成します。

ワークフロー名を設定します。
サービスアカウントを設定します。このアカウントが、ワークフローを実行するアカウントです。今回定義するワークフローコードを実行するのに必要な権限を持ったアカウントを使用します。必要な権限は、単一 CSV ファイルのインポートを参照してください。
トリガーを設定します。Workflows のトリガー設定から Eventarc のトリガーを新規に作成することができます。
1. トリガー名を設定しください。
2. イベントプロバイダとして Cloud Storage を選択してください。
3. イベントとして、finalized を選択します。このイベントは Cloud Storage へ Object の更新が完了したときに生成されます。
4. バケットでは、GCS の準備で作成したものを選択してください。
処理を記述します。ワークフローコードは、単一 CSV ファイルのインポートの内容を踏襲してください。

単一 CSV ファイルのインポート

Google Cloud Storage（GCS）上にある単一の CSV ファイルを Cloud SQL にインポートするには、Cloud SQL Admin API のインポート用 API を呼び出せばよいです。これは、以下のようなワークフローコードで実行できます。

- callImport:
    call: http.post
    args:
      url: ${"https://sqladmin.googleapis.com/v1/projects/" + projectid + "/instances/" + instance + "/import"}
      auth:
        type: OAuth2 # WorkflowがAPIコールに認証ヘッダーを追加できるようにするために必要
      body:
        importContext:
          uri: ${file}
          database: ${databaseschema}
          fileType: CSV
          csvImportOptions:
            table: ${importtable}

ただし、このワークフローを実行するサービスアカウントは、Cloud SQL の管理者でなければなりません。

また、この操作は、非同期であり、かつ、非同期処理用のコネクタが用意されていないため、以下のような対処が必要になります。

インポート操作の状態を定期的に確認することで、インポートの終了を確認します。
ジョブが終了したら、他のファイルを反復処理するか、終了するかのどちらかの操作を行います。

以上を踏まえると、 CSV ファイルを Cloud SQL にインポートするためのワークフローコードは以下のように記述できます。

import_file:
  params:
    - projectid
    - instance
    - databaseschema
    - importtable
    - file
  steps:
    - callImport:
        call: http.post
        args:
          url: ${"https://sqladmin.googleapis.com/v1/projects/" + projectid + "/instances/" + instance + "/import"}
          auth:
            type: OAuth2
          body:
            importContext:
              uri: ${file}
              database: ${databaseschema}
              fileType: CSV
              csvImportOptions:
                table: ${importtable}
        result: operation
    - chekoperation:
        switch:
          - condition: ${operation.body.status != "DONE"}
            next: wait
        next: completed
    - completed:
        return: "done"
    - wait:
        call: sys.sleep
        args:
          seconds: 1
        next: getoperation
    - getoperation:
        call: http.get
        args:
          url: ${operation.body.selfLink}
          auth:
            type: OAuth2
        result: operation
        next: chekoperation

参考

備考

Hakky ではエンジニアを募集中です！まずは話してみたいなどでも構いませんので、ぜひお気軽に採用ページからお問い合わせくださいませ。