Amazon AppFlowをGA4に接続し、人気記事ランキングをSlackに自動投稿する

こんにちは、ENECHANGE株式会社でVPoTを務めている岩本 (iwamot) です。

このブログの人気記事ランキングをSlackに自動投稿するよう実装したので、事例としてご紹介します。Google Analytics 4 (GA4) のレポートを見に行かなくても、どんな記事が読まれているのか手軽に確認できて便利です。

実装方法はいろいろ考えられますが、今回選んだのは下記でした。

Amazon AppFlowを使い、GA4データをAmazon S3に保存
保存をトリガーにAWS Lambda関数を呼び出し、データを加工してSlackに投稿

以下、AppFlowを中心に詳しくご説明します。

SaaSとAWSのデータ連携が自動化できるAmazon AppFlow

AppFlowは、SaaSとAWSサービス間のデータ連携を自動化できるサービスです。たとえば、Salesforce・SAP・Google Analyticsなどのデータを、Redshift・S3などに連携できます。

左側のSaaSから、右側のAWSサービスへ、AppFlowでデータ連携できる
（画像引用元：https://aws.amazon.com/jp/appflow/）

ENECHANGEでは、ほとんどのワークロードをAWSで運用しているので、人気記事ランキングの自動投稿もAWSで運用したいところです。

そこで、AppFlowを使って、GA4のデータをS3に保存することにしました。

事前準備：AppFlowとGA4の接続を作成

AppFlowでGA4のデータをS3に保存するには、まず接続を作成する必要があります。ぼくは、AWSのドキュメントや、クラスメソッドさんの記事を参考に進めました。

docs.aws.amazon.com

dev.classmethod.jp

AppFlowのフローを作成

続いて、AppFlowのフローを作成します。こちらは、AWSのソリューションアーキテクト・はまーんさんの記事を参考に進めました。

track3jyo.com

具体的な設定は下記の通りです。

送信元の設定

送信元として、GA4の接続やオブジェクトを選びます。

送信先の設定

送信先として、S3バケットを選びます。

ファイル形式はCSVにしました。デフォルトのJSON Linesデータより、今回のLambda関数では処理しやすいためです。

フロートリガーの設定

フロートリガーは月曜10時としました。

転送モードを増分転送にしたのは、週間ランキングを投稿したいからです。完全転送だと月間（30日）ランキングになります。

フィールドマッピングの設定

フィールドマッピングでは下記を指定しました。

customEvent:post_date（投稿日。はてなブログ独自のカスタムディメンション）
screenPageViews（ページビュー数）
unifiedPagePathScreen（URLのパス。例：/entry/2022/10/25/100000）
unifiedScreenName（記事名）

これで下記のようなCSVが得られます。

"unifiedScreenName","unifiedPagePathScreen","screenPageViews","customEvent:post_date"
"DockerfileでのENTRYPOINTとCMDの関係を整理する 〜 2つの記述形式に環境変数展開とシェル引数展開を添えて 〜 - ENECHANGE Developer Blog","/entry/2022/10/25/100000","40","2022-10-25"

パーティションと集約の設定

パーティションの設定は「毎日」としました。S3のフォルダパスを見るだけで、何年何月何日に作られたデータか分かるからです。

集約の設定は、1ファイルとしました。複数ファイルだと、Lambdaで処理しづらくなります。

フィルターの設定

フィルターは下記の2件を設定しました。

unifiedPagePathScreen が次を含む /entry/（記事ページのみに絞りたい）
unifiedScreenName が次を含む - ENECHANGE Developer Blog（翻訳されたアクセスは無視したい）

Lambda関数を作成

あとは、S3保存をトリガーとするLambda関数を作るだけです。データ加工と、Slack投稿を担当します。

はまーんさんの記事ではデータ加工にAWS Glue DataBrewを使っていますが、今回のようにLambda関数でSlackに投稿するのであれば、加工も同じ関数に任せるのがシンプルだと判断しました。

ChatGPTに相談しながら最終的にできたのが下記のコードです。

import csv
import io
import json
import os
import urllib.parse
import urllib.request

import boto3

base_url = "https://tech.enechange.co.jp"
slack_webhook_url = os.environ["SLACK_WEBHOOK_URL"]

s3 = boto3.client("s3")


def lambda_handler(event, context):

    # S3のCSVデータを取得する
    bucket = event["Records"][0]["s3"]["bucket"]["name"]
    key = urllib.parse.unquote_plus(
        event["Records"][0]["s3"]["object"]["key"], encoding="utf-8"
    )
    response = s3.get_object(Bucket=bucket, Key=key)
    body = response["Body"].read().decode("utf-8")
    csv_string = io.StringIO(body)
    reader = csv.reader(csv_string)

    # ヘッダを読み飛ばす
    _ = next(reader)

    # ページビュー数の降順でソートする
    data_rows = [row for row in reader]
    sorted_data = sorted(data_rows, key=lambda x: int(x[2]), reverse=True)

    # 上位10件のデータで投稿メッセージを組み立てる
    message_lines = []
    for i, row in enumerate(sorted_data[:10]):

        # 記事名からブログ名を削除する
        title = row[0].replace(" - ENECHANGE Developer Blog", "")

        # URLを「https://」からに整える
        url = f"{base_url}{row[1]}"

        # メッセージに行を追加する
        page_views = row[2]
        post_date = row[3]
        line = f"*{post_date}* <{url}|{title}> {page_views} views"
        message_lines.append(line)

    # メッセージを整えてSlackに投稿する
    header = "ENECHANGE Developer Blogの先週の人気記事です :medetai-nya:"
    message = header + "\n\n" + "\n".join(message_lines)
    data = json.dumps({"text": message}).encode("utf-8")
    req = urllib.request.Request(
        slack_webhook_url, data, headers={"Content-Type": "application/json"}
    )
    urllib.request.urlopen(req)

    return {
        "statusCode": 200,
        "body": "OK",
    }