Embulkを使ってMySQLのデータをBigQueryに流す手順 – データ分析入門シリーズ

Embulkを使ってMySQLのテーブルをBigQueryに流すまでの手順を書いています。

環境を準備する
プラグインをインストールする
サービスアカウントがBigQueryを操作できるようにする
1. JSONキーを取得する
2. サービスアカウントに権限を付与する
設定ファイルを書く
実行する
トラブルシューティング
1. Permission bigquery.datasets.get denied
2. User does not have bigquery.jobs.create permission

環境を準備する

MySQLが動いているサーバ（あるいはMySQLにアクセス可能なサーバ）にembulkをインストールしてコマンドを使えるようにしておきます。

Embulkをインストールする手順 - データ分析入門シリーズ

準備編embulkにはJavaが必要です。今回用意していたサーバにはJavaを準備していなかったので下記の手順でJavaを使えるようにしました。インストール手順公式ページ（）のクイックスタートに従って進めていきます。...

Cloud SDKのbqコマンドも使えるようにしておきます。クライアントを使わずにブラウザから操作しても良いですが、この記事ではコマンドを使っています。

GCPのBigQueryを試してみた - データ分析入門シリーズ

Google Cloud Platformも大して使ったことがなかったのに BigQuery を使ってみることになった。調べたり手を動かした結果を記録してます。BigQueryって何？大きなデータを扱うためには適切な物理サーバを用意...

プラグインをインストールする

EmbulkでMySQLとBigQueryを扱えるようにプラグインをインストールします。下記の2つのコマンドを実行するとインストールできます。

$ embulk gem install embulk-input-mysql
...
Fetching: embulk-input-mysql-0.12.3-java.gem (100%)
Successfully installed embulk-input-mysql-0.12.3-java
1 gem installed

$ embulk gem install embulk-output-bigquery
...
Fetching: embulk-output-bigquery-0.6.7.gem (100%)
Successfully installed embulk-output-bigquery-0.6.7
22 gems installed

サービスアカウントがBigQueryを操作できるようにする

BigQuery APIが有効になっていればサービスアカウントがあります。embulkでバルクインサートする際はこのサービスアカウントを使うので、サービスアカウントが各操作をできるように準備します。

JSONキーを取得する

Google Cloud Platformの画面、左メニューから「IAMと管理」→「サービスアカウント」と進みます。前回までの手順でサービスアカウントが作成されていて、リストにアカウントがあればOKです。もし存在していなければ「サービスアカウントを作成」から新規作成することができます。

次に、使用するサービスアカウントの「操作」から「鍵を管理」を選びます（サービスアカウントを選択して、遷移先のページで「キー」タブを選択しても同じです）。

「鍵を追加」から「新しい鍵を作成」を選択します。

「キーのタイプ」で「JSON」を選択して「作成」ボタンを押すと、鍵が作成とダウンロードが始まります。この鍵は大切に保存しておいてください。

サービスアカウントに権限を付与する

サービスアカウントが諸々の操作をできるように権限を付与します。

左メニューから「IAMと管理」→「IAM」と進みます。

上部の「追加」ボタンを押してプリンシパルを追加する画面に移ります。

ここで「新しいプリンシパル」にサービスアカウントのメールアドレスを入力します。

次にロールを追加します。「ロールを選択」のタブから「BigQuery データ編集者」と「BigQuery ジョブユーザー」の二つのロールを選択します。ロールはフィルタを使って検索することができます。最後に保存を押して完了です。

「IAM」の画面で、サービスアカウントの行に指定したロールが付与されていればOKです。

設定ファイルを書く

ID、名前、性別を持ったユーザテーブルをBigQueryに流す想定です。前回までの手順でBigQuery側にはテーブルが用意されている想定です。

in:
  type: mysql
  host: (ホスト名)
  user: (ユーザ名)
  password: (パスワード)
  database: (データベース名)
  table: users (=テーブル名)
  select: user_id, user_name, user_gender (=カラム名)
  options: {useLegacyDatetimeCode: false, serverTimezone: Asia/Tokyo}

入力側です。

out:
  type: bigquery
  auth_method: service_account
  service_account_email: (サービスアカウント名)
  json_keyfile: (JSONキーのパス)
  path_prefix: /tmp/
  file_ext: .csv.gz
  source_format: CSV
  project: (プロジェクト名)
  dataset: (データセット名)
  auto_create_table: true
  table: (テーブル名)
  formatter: {type: csv, charset: UTF-8, delimiter: ',', header_line: false}
  encoders:
  - {type: gzip}

出力側です。

inとoutをまとめてuser.ymlとして保存します。

実行する

こちらの記事のサンプルで動かしたのと同じ要領で実行します。

まずはpreviewでdry-runしてみます。

$ embulk preview user.yml
...
2021-10-07 11:43:41.879 +0900 [INFO] (0001:preview): Fetch size is 10000. Using server-side prepared statement.
2021-10-07 11:43:41.879 +0900 [INFO] (0001:preview): Connecting to jdbc:mysql:...
2021-10-07 11:43:41.902 +0900 [INFO] (0001:preview): SQL: SELECT user_id, user_name, user_gender FROM `members`                                  
2021-10-07 11:43:42.029 +0900 [INFO] (0001:preview): > 0.12 seconds
...

想定通りのSQLクエリが実行されていることが確認できました。上記では省略していますが、標準出力に対象のテーブルのレコードが数行表示されると思います。

次は本実行です。

$ embulk run user.yml

サブコマンドを変えるだけです。previewのときは序盤のデータだけ処理されましたが、runではさらに多くのデータが処理される様子がログに表示されます。一時ファイルが作られていることもログから読み取ることができます。

無事に実行が完了したらBigQuery側で確認してみます。

トラブルシューティング

Permission bigquery.datasets.get denied

bigquery上のデータセットのメタデータを取得できないエラーです。サービスアカウントのロール設定で適切なロールを設定してください。

org.embulk.exec.PartialExecutionException: org.jruby.exceptions.RaiseException: (Error) failed to get dataset PROJECT_NAME:DATASET_NAME, response:{:status_code=>403, :message=>”accessDenied: Access Denied: Dataset PROJECT_NAME:DATASET_NAME: Permission bigquery.datasets.get denied on dataset PROJECT_NAME:DATASET_NAME (or it may not exist).”, :error_class=>Google::Apis::ClientError}

ちなみに、「bigquery.datasets.get」はメタデータを取得するためのパーミッションですが、他にどんなものがあるか、どういうパーミッションなのかはこちらで確認することができます