Amazon AppFlow の BigQuery コネクタを試す

## はじめに
[Amazon AppFlow](https://aws.amazon.com/jp/appflow/) はさまざまな SaaS アプリケーションや AWS サービス間でデータを簡単に転送できるサービスです。

AppFlow でサポートされるアプリケーションは順次追加されており、AWS サービスも含めると約 80 のコネクタが提供されています。また [Custom Connector SDK](https://aws.amazon.com/jp/about-aws/whats-new/2022/02/amazon-appflow-launches-custom-connector-sdk/) を使用して独自のコネクタを開発することもできます。

2023/6/15 のアップデートで Amazon AppFlow に Google BigQuery コネクタが追加されました。

https://aws.amazon.com/jp/about-aws/whats-new/2023/06/amazon-appflow-new-data-connectors/

データソース、データの宛先、どちらのユースケースにも対応しています

* データソースとして BigQuery から他のサービスにデータを転送する
* データの宛先としてデータを他のサービスから BigQuery へ転送する

## やってみる
シンプルなパターンとして Google BigQuery から Amazon S3 へのデータフローを作成してみます。

:::note info
BigQuery をデータソースとする場合、宛先として指定できるサービスには制限がありますのでご注意ください。
:::

https://docs.aws.amazon.com/appflow/latest/userguide/connectors-googlebigquery.html#googlebigquery-destinations

> When you create a flow that uses Google BigQuery as the data source, you can set the destination to any of the following connectors:
>
> * Amazon Honeycode
> * Amazon Lookout for Metrics
> * Amazon Redshift
> * Amazon RDS for PostgreSQL
> * Amazon S3
> * HubSpot
> * Marketo
> * Salesforce
> *  SAP OData
> * Snowflake
> * Upsolver
> * Zendesk
> * Zoho CRM

### データの準備
BigQuery 側に以下のようなテーブルとデータを準備しました。

:::note info
詳細は後述しますが、2023 年 7 月時点で今回のサンプルデータの ADDRESS フィールドのようなネストされたデータの取得は未サポートでした。
:::

**スキーマ**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/35d097df-0893-b175-fec7-4daa51cf656e.png)
**データ**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/aada8bd2-e7ed-aca5-f6b0-0b8adb936704.png)


### OAuth クライアントの作成
AppFlow が BigQeury にアクセスできるようにするため、事前に Google Cloud 側で OAuth クライアント ID を作成しておく必要があります。

Google Cloud コンソールの「API とサービス」から認証情報へ進み、OAuth クライアント ID を新規作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/6d52164e-18a7-3a1a-591d-b514613735ad.png)

以下を設定し、OAuth クライアント ID を作成します。
* アプリケーションの種類はウェブ アプリケーションとし、任意の名前を入力します
* 承認済みの JavaScript 生成元には `https://console.aws.amazon.com` を入力します
* 承認済みのリダイレクト URI には `https://<region>.console.aws.amazon.com/appflow/oauth` を設定します
    * `<region>` は使用する AppFlow のリージョンコードに書き換えてください

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/4a483d46-a878-53d6-2a79-afa31749c089.png)

OAuth クライアント ID には適切な Scope を追加する必要があります。以下のドキュメントを参考に BigQuery API v2 のアクセスに必要な Scope を追加します。

https://developers.google.com/identity/protocols/oauth2/scopes?hl=ja#bigquery

| スコープ | 説明 |
| --- | --- |
| https\://www.googleapis.com/auth/bigquery | View and manage your data in Google BigQuery and see the email address for your Google Account |
| https\://www.googleapis.com/auth/bigquery.insertdata | Insert data into Google BigQuery |
| https\://www.googleapis.com/auth/cloud-platform | See, edit, configure, and delete your Google Cloud data and see the email address for your Google Account. |
https:\//www.googleapis.com/auth/cloud-platform.read-only | View your data across Google Cloud services and see the email address of your Google Account |
https:\//www.googleapis.com/auth/devstorage.full_control | Manage your data and permissions in Cloud Storage and see the email address for your Google Account |
https\://www.googleapis.com/auth/devstorage.read_only |	View your data in Google Cloud Storage |
https\://www.googleapis.com/auth/devstorage.read_write | Manage your data in Cloud Storage and see the email address of your Google Account |

API とサービスの OAuth 同意画面からスコープを追加します。今回は検証のためすべてのスコープを追加したのですが、BigQuery を宛先としてデータの書き込みを行わない場合は Write 系の Scope は不要と思われます。


### 接続の作成
AppFlow コンソールの接続で Google BigQuery コネクタを選択し、接続の作成をクリックします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/1dc48c04-858e-f51f-7602-87f3acda858a.png)

以下の項目を設定し、新規の接続を作成します。

* 任意の接続名を入力します
* access_type は `offline` のみ選択できます
* 作成した OAuth クライアントのクライアント ID およびクライアントシークレットを入力します
* 必要に応じて KMS キーによるデータ暗号化を設定することもできます

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/87f1b8ae-0328-9fb7-e35c-6412d9a66d5b.png)

接続するをクリックすると、Google の画面に遷移しますのでアクセスリクエストを許可します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/8e2a947f-1a7c-7c10-7625-54c020502428.png)

正常に接続が作成できたらフローの作成に進みます。

### フローの作成
AppFlow コンソールのフローから新規のフローを作成します。

フローの詳細で任意のフロー名と説明を入力します。必要に応じて KMS キーによるデータ暗号化やタグを設定することもできます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/5fa989e4-7e75-ea7d-7590-6772b7b80bb3.png)

送信元の詳細を以下のように設定します

* 送信元名で Google BigQuery コネクタを選択します
* 接続の選択で先ほど作成した接続を選択します
* オブジェクトの選択では `Table` のみ選択できます
* サブオブジェクトの指定で接続するプロジェクト、データセット、テーブル名をそれぞれ選択します
* Data transfer settings はデフォルトの `Automatic` のままとしました

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/249c4459-02f4-0995-971c-7823a6e4c49b.png)

送信先の詳細で Amazon S3 を選択し、宛先となる S3 バケットの情報を入力します。オプションで AWS Glue Data Catalog へのテーブル登録やファイル形式の設定を行うことができます。今回はデフォルトのままデータカタログへの登録なし、JSON 形式での出力としています。

検証目的でスケジュール実行する必要がないため、フロートリガーはオンデマンドを選択しました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/1eab1c9f-e9bc-8606-66ad-cec32ede5dcc.png)

データフィールドのマッピングではすべてのフィールドを直接マッピングしています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/ba17b8ff-8b5b-9cbe-8e56-bdc8c45eedb3.png)

パーティションを集約の設定はお好みで選択し、次に進みます。今回は実行 ID をパーティションとし、集約の設定は行いませんでした。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/ea681322-166f-9af3-ec1e-df603bd3d232.png)

AppFlow ではデータフィールドにフィルターを追加することでフィルター条件を満たすレコードのみを転送することができますが、今回は特に追加せずそのまま次へ進みます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/6b0b4f4e-05b3-a07f-c2c6-cb705e90e93f.png)

確認画面で設定内容を確認し、フローの作成を完了します。


### フローの実行とトラブルシューティング
動作確認を行うため。作成済みのフローからフローの実行をクリックします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/0dcd7321-3203-2284-7f14-061a054efa52.png)

少し待つと、フローの実行がエラーになってしましました:angel_tone2:

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/d3fd138e-f840-8803-7dd9-1a321b5d849e.png)

```
Amazon AppFlow received the following error:
The request failed because the service Source Google BigQuery returned the following error: Details: The request failed with status code 400 (Bad Request)., ErrorCode: ClientError.
```

AppFlow のエラーメッセージでは BigQuery 側からステータスコード 400 (Bad Request) が返り、リクエストが失敗したことしかわかりませんでした。このような場合の原因調査は Google Cloud 側で行う必要があります。

Cloud Logging で Big Query API 実行時のエラー詳細が確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/4bd1fcf6-3f51-3f5e-d8dd-be5e05aa1599.png)

```
Field <フィールド名> from table <テーブル名> is not a leaf field. Consider using  SQL, which allows selecting non-leaf fields.
```

上記のエラーメッセージをもとに AWS Support に問い合わせをおこなったところ、2023 年 7 月時点でネストされたデータの取得は未サポートということがわかりました。

そのためフローを編集し、ADDRESS フィールドのマッピングを削除した上で再度実行してみます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/25b4e19d-9d12-a884-f849-bfc0839baa3f.png)

今度は正常に実行されました！

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/f9475b0d-acc6-3d19-f5cf-9cf490204173.png)

S3 バケットにファイルが出力されていることが確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/3bd0fa22-1d6b-bd87-bd0a-3f5a760e65ca.png)

冒頭で準備したデータが BigQuery から JSON 型式で取得できていました。

```result.json
{"FIRST_NAME":"Jeff","LAST_NAME":"Smith","DATE_OF_BIRTH":"1980-10-10"}
{"FIRST_NAME":"Charlotte","LAST_NAME":"Lalande","DATE_OF_BIRTH":"1990-01-01"}
```

## まとめ
AppFlow に BigQuery コネクタが追加されたことにより、BigQuery とのデータ連携が簡単に実装できるようになりました。対応するサービス間で双方向のフローを作成できるのは大きなメリットだと感じます。AWS 側から Google BigQuery のデータを取得するという観点では、これまでも Amazon Athena でデータソースコネクタが提供されていましたが、より手軽さや柔軟性が増したのではないかと思います。

## 参考
https://docs.aws.amazon.com/appflow/latest/userguide/connectors-googlebigquery.html

https://docs.aws.amazon.com/athena/latest/ug/connectors-bigquery.html

以上です。
参考になれば幸いです。