Amazon AppFlow と Amazon Athena を使用して Google Analytics データを分析する

Software as a Service (SaaS) アプリケーションの重要性は急速に高まってきています。このデータは、ビジネスの意思決定に影響を与える分析を実行するときに含めることが不可欠です。Amazon AppFlow は、SaaS データをデータレイクに安全に転送するのを助けるフルマネージド統合サービスです。データ転送フローは、オンデマンドで、スケジュールに従い、またはイベント発生後に実行できます。Amazon Athena を使用してこのデータをすばやく分析し、Amazon Simple Storage Service (Amazon S3) に既に保存されている多数のデータセットと結合できます。複数の SaaS データセットを結合し、さらに Athena 横串検索機能を使って、Amazon Relational Database Service (Amazon RDS) などの従来のデータベースにある運用データと結合できます。

この記事では、Amazon AppFlow を使用して Google Analytics データを抽出し、それを Amazon S3 に保存して、Athena でクエリできるようにする方法を説明します。

アーキテクチャの概要

次の図は、この記事で説明するフローを示しています。最初に Amazon AppFlow 内に新しいフローを作成して、Google Analytics データを Amazon S3 に転送します。転送するデータの形式は、Athena がサポートしていない複数行の JSON です。AWS Lambda 関数は、この JSON 形式のファイルを Apache Parquet 形式に変換します。この変換により、クエリを効率的かつコスト効率よく実行できます。この関数には、Amazon S3 プレフィックスの変更や Hive スタイルのパーティションを使用したデータの保存など、他の変換を含めることもできます。Amazon AppFlow は、新しいデータのみを抽出するスケジュールされたジョブをサポートしています。そのため、Amazon S3 イベントトリガーと変換 Lambda 関数を使用して自動ワークフローを開発できます。Amazon AppFlow は現在 15 のリージョンでご利用いただけます。S3 バケットが配置されているリージョンを選択してください。このチュートリアルでは、米国東部 (バージニア北部) を使用します。

この記事では、サンプルの Google アカウント、適切なアクセス権限を持つ OAuth クライアント、および Google アナリティクスデータを使用します。独自の Google リソースを使用することもできます。Amazon AppFlow からの Google Analytics アクセスを有効にするには、事前に新しい OAuth クライアントをセットアップする必要があります。次の手順を実行します。

Google API コンソール (https://console.developers.google.com) で、[Library] を選択します。
検索フィールドに「analytics」と入力します。
[Google Analytics API] を選択します。
[ENABLE] をクリックして、前のページに戻ります。
検索結果に表示されている [Google Analytics Reporting API] を選択します。
[ENABLE] をクリックして、メインページに戻ります。
[OAuth consent screen] を選択します。
新しい [Internal] アプリを作成します (個人アカウントを使用している場合は、[External] を選択します)。
com を [Authorized domains] として追加します。
[Add scope] を選択します。
../ auth/analytics.readonly を [Scopes for Google APIs] として追加します。
[Save] をクリックします。
[Credentials] を選択します。
OAuth クライアント ID 認証情報を追加します。
[Web application] を選択します。
承認済みの JavaScript オリジン URL として「https://console.aws.amazon.com/」と入力します。
承認済みリダイレクト URL として「https://AWSREGION.console.aws.amazon.com/appflow/oauth」と入力します。(AWSREGION を作業中のリージョンに置き換えます。us-east-1 で Amazon AppFlow を使用している場合は、「https://console.aws.amazon.com/appflow/oauth」と入力します)。
[Save] をクリックします。

Lambda と Amazon S3 のセットアップ

最初に、Amazon AppFlow 転送先として新しい S3 バケットを作成する必要があります。次に、pandas と pyarrow モジュールを使用して JSON 形式のデータを Parquet 形式に変換する新しい Lambda 関数を開発します。最後に、新しい Amazon S3 オブジェクトが作成されたときに Lambda 関数を自動的に呼び出すように Amazon S3 イベントトリガーを設定します。

新しい S3 バケットを作成する

Amazon S3 バケットを作成するには、次の手順を実行します。

Amazon S3 コンソールで、[バケットの作成] を選択します。
バケットの名前を入力します。例: appflow-ga-sample。
[バケットの作成] を選択します。

Lambda レイヤー用の.zip ファイルを準備する

pandas と pyarrow モジュールを含む.zip ファイルを作成するには、次の手順を実行します。

Docker を実行できる環境をセットアップします。
次のコマンドを実行します。

mkdir python
docker run -it --rm -v $(pwd)/python:/python python:3.6.8 pip install -t /python pandas==0.23.4 pyarrow==0.11.1
zip -r pandas-pyarrow.zip python

Amazon S3 コンソールで、[appflow-ga-sample] を選択します。
[フォルダの作成] を選択します。
フォルダの名前を入力してください。例: lambda-layer。
[Save] をクリックします。
[lambda-layer] を選択します。
[アップロード] をクリックします。
[pandas-pyarrow.zip] を選択し、[アップロード] をクリックします。

Parquet エクスポート用の Lambda レイヤーを作成する

Lambda レイヤーを作成するには、次の手順を実行します。

Lambda コンソールで、[レイヤー] を選択します。
[レイヤーの作成] を選択します。
[名前] には、レイヤーの名前を入力します。例: pandas-parquet。
[Amazon S3 からファイルをアップロードする] を選択します。
[Amazon S3 のリンク URL] には、zip ファイルの Amazon S3 パスを入力します。例: s3://appflow-sample/lambda-layer/pandas-parquet.zip。
[互換性のあるランタイム] では、[Python 3.6] を選択します。
[作成] をクリックします。

データ変換用の Lambda 関数を作成する

Lambda 関数を作成して Amazon S3 イベントをトリガーするには、次の手順を実行します。

Lambda コンソールで、[関数の作成] を選択します。
[一から作成] を選択します。
[関数名] には、関数の名前を入力します。例: ga-converter。
[ランタイム] には、[Python 3.6] を選択します。
[基本的な Lambda アクセス権限で新しいロールを作成] を選択します。
[関数の作成] を選択します。

Lambda 関数設定で、[関数コード] の [lambda_function] エリアに次のコードを入力します

この Lambda 関数は、AppFlow 出力ファイルをダウンロードし、Google アナリティクスの JSON ファイルから必要なデータを抽出して、Parquet 形式に変換します。最後に、別のキー名で再度 Amazon S3 にアップロードします。必要に応じて、スクリプトを変更できます (特に、ディメンションや値の名前、またはその他の変換)。

import pandas as pd
import boto3
from datetime import datetime as dt

def lambda_handler(event, context):
    bucket_name = event['Records'][0]['s3']['bucket']['name']
    object_key = event['Records'][0]['s3']['object']['key']

    s3_client = boto3.client('s3')
    raw_object = s3_client.get_object(Bucket=bucket_name, Key=object_key)
    raw_data = json.loads(raw_object['Body'].read().decode('utf-8'))
    
    record_dates = [dt.strptime(r['dimensions'][0], '%Y%m%d%H') for r in raw_data['reports'][0]['data']['rows']]
    devices = [r['dimensions'][1] for r in raw_data['reports'][0]['data']['rows']]
    user_counts = [int(r['metrics'][0]['values'][0]) for r in raw_data['reports'][0]['data']['rows']]
    df = pd.DataFrame({
        'year': [r.year for r in record_dates],
        'month': [r.month for r in record_dates],
        'day': [r.day for r in record_dates],
        'hour': [r.hour for r in record_dates],
        'device': devices,
        'user_count': user_counts
    })
    
    output_file = dt.now().strftime('%Y%m%d%H%M%S')
    output_path = '/tmp/{}.parquet'.format(output_file)
    df.to_parquet(output_path)

    s3_resource = boto3.resource('s3')
    bucket = s3_resource.Bucket(bucket_name)
    bucket.upload_file(output_path, 'ga-data/{}.parquet'.format(output_file))

レイヤーを追加する

Lambda 関数にレイヤーを追加するには、次の手順を実行します。

[デザイナー] で [レイヤー] を選択します
[レイヤーの追加] を選択します。
[ランタイムと互換性のあるレイヤーのリストから選択] を選択します。
[名前] には、pandas-pyarrow を選択します。
[バージョン] では、[1] を選択します。
[追加] を選択します。

タイムアウト設定を増やす

Lambda タイムアウト設定を増やすには、次の手順を実行します。

[設定] タブの [基本設定] で、[編集] を選択します。
30 秒を [タイムアウト] として設定します。
[Save] をクリックします。
[設定] タブで、[保存] を選択します。

イベントトリガーを追加する

Amazon S3 イベントトリガーを追加するには、次の手順を実行します。

[デザイナー] セクションで、[トリガーの追加] を選択します。
[S3] を選択し、作成したバケットを選択します。
[イベントタイプ] では、[すべてのオブジェクト作成イベント] を選択します。
[プレフィックス] には、「raw」と入力します。
[追加] を選択します。

アクセス権限を追加する

この Lambda 関数が Amazon S3 オブジェクトを読み書きするための適切なアクセス権限を追加するには、次の手順を実行します。

[アクセス権限] タブで、ロール名を入力します。例: ga-converter-role-zm6u0f4g。
AWS Identity and Access Management (IAM) コンソールで、[ポリシー] を選択します。
[ポリシーの作成] を選択します。
[JSON] タブで、次のポリシーを入力します (ターゲットバケット名 arn:aws:s3:::appflow-ga-sample* を独自のバケット名に置き換えます)。

{	
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject"
             ],
            "Resource": [
                "arn:aws:s3:::appflow-ga-sample*"
            ]
        }
    ]
}

[ポリシーの確認] を選択します。
新しいポリシーの名前を入力します。例: lambda-s3-ga-converter-policy。
[ポリシーの作成] を選択します。
IAM コンソールで、以下を選択します
検索フィールドにロール名 (ga-converter-role-zm6u0f4g) を入力します。
ロールを選択します。
[ポリシーのアタッチ] を選択します。
[lambda-s3-ga-converter-policy] を選択します。
[ポリシーのアタッチ] を選択します。

Amazon AppFlow のセットアップ

これで、Google アナリティクスから Amazon S3 に転送する新しい Amazon AppFlow フローを作成できます。新しい Amazon AppFlow 転送フローを作成するには、次の手順を実行します。

Amazon AppFlow コンソールで、[フローを作成] を選択します。
フローの名前を入力します。例: my-ga-flow。
[次へ] をクリックします。
[ソース名] では、[Google Analytics] を選択します。
[新規接続を作成] を選択します。
OAuth クライアント ID とクライアントシークレットを入力してから、接続に名前を付けます。例: ga-connection。
ポップアップウィンドウで、amazon.com に Google Analytics API へのアクセスを許可することを選択します。
[Google Analytics オブジェクトの選択] では、[レポート] を選択します。
[Google Analytics ビューの選択] では、[すべてのウェブサイトデータ] を選択します。
[送信先名] では、[Amazon S3] を選択します。
[バケットの詳細] では、作成したバケットを選択します。
プレフィックスとして「raw」と入力します。
[オンデマンドで実行] を選択します。

[次へ] をクリックします。
[手動でフィールドをマッピングする] を選択します。
[ソースフィールド名] では次の 3 つのフィールドを選択します。
- Time: DIMENSION: ga:dateHour
- Platform or Device: DIMENSION: ga:deviceCategory
- User: METRIC: ga:users
[フィールドを直接マッピングする] を選択します。

[次へ] をクリックします。
[フィルターを追加] セクションで、[次へ] を選択します。
[フローを作成] を選択します。

フローを実行する

新しいフローを作成したら、オンデマンドで実行できます。

Amazon AppFlow コンソールで、[my-ga-flow] を選択します。
[フローを実行] を選択します。

このチュートリアルでは、理解しやすいようにオンデマンドジョブの実行を選択します。実際には、スケジュールされたジョブを選択して、新しく追加されたデータのみを定期的に抽出できます。Amazon S3 イベントトリガーは、データを自動的に変換するのにも役立ちます。

Athena によるクエリ

クエリを実行する前に、外部テーブルを作成する必要があります。次の手順を実行します。

Athena コンソールで、クエリエディタに「create database appflow_data」と入力します。
[クエリの実行] を選択します。
クエリエディタで次のコマンドを入力します (ターゲットバケット名 appflow-ga-sample を独自のバケットに置き換えます)。

CREATE EXTERNAL TABLE appflow_data.ga_sample (
  `year` int,
  `month` int,
  `day` int,
  `hour` int,
  `device` string,
  `user_count` int
 )
STORED AS PARQUET
LOCATION 's3://appflow-ga-sample/ga-data'
tblproperties ("parquet.compression"="SNAPPY")
;

[クエリの実行] を選択します。

これで、Google Analytics データをクエリできます。次のクエリを入力して実行します。このクエリは、1 時間ごとにユーザーのウェブサイトにアクセスするのに一般的に使用されているデバイスの種類を示します。

SELECT
  year
  , month
  , day
  , device
  , count(user_count) as cnt
FROM
  appflow_data.ga_sample
GROUP BY
  year
  , month
  , day
  , device
ORDER BY
  cnt DESC
LIMIT 10
;

次のスクリーンショットは、クエリ結果を示しています。

まとめ

この記事では、Amazon AppFlow を使用して Google Analytics データを Amazon S3 に転送し、Amazon Athena で分析する方法を示しました。Google Analytics やその他の SaaS アプリケーションからデータを抽出するのに独自のアプリケーションを構築する必要がなくなりました。Amazon AppFlow を使用すると、完全に自動化されたデータ転送、変換ワークフローおよび統合クエリ環境を一元的に開発できます。

著者について

志村誠は、アマゾンウェブサービスの分析分野のスペシャリストソリューションアーキテクトです。彼は、お客様が AWS プラットフォームで効率的なデータパイプラインを開発するのをサポートしています。以前は、データエンジニアとして、分散データプラットフォームを開発していました。仕事以外では、家族と過ごしたり、犬と遊んだり、ビデオゲームを楽しんだりしています。

Amazon Web Services ブログ