AWS DataOps 개발 키트를 사용하여 Google 애널리틱스 데이터를 수집, 변환, 분석할 수 있는 데이터 파이프라인을 구축하세요.

작성자: 안톤 쿠쿠스킨 (AWS) 과 루디 푸이그 () AWS

요약

이 패턴은 AWS DataOps Development Kit (DDK) 및 기타 서비스를 사용하여 Google 애널리틱스 데이터를 수집, 변환 및 분석하기 위한 데이터 파이프라인을 구축하는 방법을 설명합니다. AWS 데이터 워크플로와 최신 데이터 아키텍처를 구축하는 데 도움이 되는 오픈 소스 개발 AWS DDK 프레임워크입니다. AWS 의 주요 목표 중 하나는 파이프라인 조정, 인프라 구축, 인프라 구축과 같은 노동 집약적인 데이터 파이프라인 작업에 일반적으로 소요되는 시간과 노력을 절약하는 것입니다. AWS DDK DevOps 이러한 노동 집약적인 작업을 AWS DDK 오프로드하여 코드 작성 및 기타 고부가가치 활동에 집중할 수 있습니다.

사전 조건 및 제한 사항

사전 조건 

제품 버전

아키텍처

기술 스택

대상 아키텍처 

다음 다이어그램은 Google Analytics 데이터를 수집, 변환, 분석하는 이벤트 기반 프로세스를 보여줍니다.

이 다이어그램은 다음 워크플로를 보여줍니다.

도구

AWS도구

코드

이 패턴의 코드는 GitHub AWS DataOps 개발 키트 (DDK) 및 Amazon AppFlow, Amazon Athena 및 개발 키트 리포지토리를 사용한 Google 애널리틱스 데이터 분석에서 확인할 수 있습니다. AWS DataOps

에픽

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

문제 해결

관련 리소스

추가 정보

AWSDDK데이터 파이프라인은 하나 이상의 단계로 구성됩니다. 다음 코드 예제에서는 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 AppFlow, Amazon Athena 및 AWS DataOps 개발 키트 리포지토리를 GitHub 사용한 Google 애널리틱스 데이터 분석을 참조하십시오.