Development AWS DataOps Kit を使用して Google Analytics データを取り込み、変換、分析するためのデータパイプラインを構築する

作成者: Anton Kukushkin (AWS) と Rudy Puig (AWS）

[概要]

このパターンでは、 AWS DataOps 開発キット (DDK) やその他の AWSサービスを使用して、Google Analytics データを取り込み、変換、分析するためのデータパイプラインを構築する方法について説明します。AWS DDK は、 でデータワークフローと最新のデータアーキテクチャを構築するのに役立つオープンソースの開発フレームワークですAWS。の主な目的の 1 AWSDDKつは、パイプラインのオーケストレーション、インフラストラクチャの構築、そのインフラストラクチャの DevOps 背後にある の作成など、通常、労働集約的なデータパイプラインタスクに専念する時間と労力を節約することです。これらの労力のかかるタスクを にオフロードAWSDDKして、コードやその他の価値の高いアクティビティの作成に集中できます。

前提条件と制限

前提条件

製品バージョン

アーキテクチャ

テクノロジースタック

ターゲット アーキテクチャ

次の図は、Google アナリティクスのデータを取り込み、変換、分析するイベント駆動型のプロセスを示しています。

この図表は、次のワークフローを示しています：

ツール

AWS ツール

コード

このパターンのコードは、 GitHub AWS DataOps 開発キット (DDK） および Amazon 、Amazon Athena AppFlow、および AWS DataOps 開発キットリポジトリを使用した Google Analytics データの分析で使用できます。

エピック

git clone https://github.com/aws-samples/aws-ddk-examples.git

cd google-analytics-data-using-appflow/python && python3 -m venv .venv

source .venv/bin/activate && pip install -r requirements.txt

トラブルシューティング

関連リソース

追加情報

AWS DDK データパイプラインは、1 つ以上のステージで構成されます。次のコード例では、AppFlowIngestionStage を使用して Google Analytics からデータを取り込み、SqsToLambdaStage を使用してデータ変換を処理し、AthenaSQLStage を使用して Athena クエリを実行します。

まず、次のコード例に示すように、データ変換ステージと取り込みステージを作成します。

        appflow_stage = AppFlowIngestionStage(
            self,
            id="appflow-stage",
            flow_name=flow.flow_name,
        )
        sqs_lambda_stage = SqsToLambdaStage(
            self,
            id="lambda-stage",
            lambda_function_props={
                "code": Code.from_asset("./ddk_app/lambda_handlers"),
                "handler": "handler.lambda_handler",
                "layers": [
                    LayerVersion.from_layer_version_arn(
                        self,
                        id="layer",
                        layer_version_arn=f"arn:aws:lambda:{self.region}:336392948345:layer:AWSDataWrangler-Python39:1",
                    )
                ],
                "runtime": Runtime.PYTHON_3_9,
            },
        )
        # Grant lambda function S3 read & write permissions
        bucket.grant_read_write(sqs_lambda_stage.function)
        # Grant Glue database & table permissions
        sqs_lambda_stage.function.add_to_role_policy(
            self._get_glue_db_iam_policy(database_name=database.database_name)
        )
        athena_stage = AthenaSQLStage(
            self,
            id="athena-sql",
            query_string=[
                (
                    "SELECT year, month, day, device, count(user_count) as cnt "
                    f"FROM {database.database_name}.ga_sample "
                    "GROUP BY year, month, day, device "
                    "ORDER BY cnt DESC "
                    "LIMIT 10; "
                )
            ],
            output_location=Location(
                bucket_name=bucket.bucket_name, object_key="query-results/"
            ),
            additional_role_policy_statements=[
                self._get_glue_db_iam_policy(database_name=database.database_name)
            ],
        )

次に、次のコード例に示すように、 DataPipelineコンストラクトを使用して、 EventBridge ルールを使用してステージをまとめて「ワイヤ」します。

        (
            DataPipeline(self, id="ingestion-pipeline")
            .add_stage(
                stage=appflow_stage,
                override_rule=Rule(
                    self,
                    "schedule-rule",
                    schedule=Schedule.rate(Duration.hours(1)),
                    targets=appflow_stage.targets,
                ),
            )
            .add_stage(
                stage=sqs_lambda_stage,
                # By default, AppFlowIngestionStage stage emits an event after the flow run finishes successfully
                # Override rule below changes that behavior to call the the stage when data lands in the bucket instead
                override_rule=Rule(
                    self,
                    "s3-object-created-rule",
                    event_pattern=EventPattern(
                        source=["aws.s3"],
                        detail={
                            "bucket": {"name": [bucket.bucket_name]},
                            "object": {"key": [{"prefix": "ga-data"}]},
                        },
                        detail_type=["Object Created"],
                    ),
                    targets=sqs_lambda_stage.targets,
                ),
            )
            .add_stage(stage=athena_stage)
        )

コード例の詳細については、「Amazon 、 GitHub Amazon Athena AppFlow、および AWS DataOps Development Kit リポジトリを使用した Google Analytics データの分析」を参照してください。