Erstellen Sie mit dem AWS DataOps Development Kit eine Datenpipeline, um Google Analytics-Daten aufzunehmen, zu transformieren und zu analysieren

Erstellt von Anton Kukushkin (AWS) und Rudy Puig () AWS

Übersicht

Dieses Muster beschreibt, wie Sie eine Datenpipeline aufbauen, um Google Analytics-Daten mithilfe des AWS DataOps Development Kit (DDK) und anderer Dienste aufzunehmen, zu transformieren und zu analysieren. AWS Das AWS DDK ist ein Open-Source-Entwicklungsframework, das Sie beim Aufbau von Datenworkflows und moderner Datenarchitektur unterstützt. AWS Eines der Hauptziele von AWS DDK ist es, Ihnen die Zeit und Mühe zu sparen, die normalerweise für arbeitsintensive Daten-Pipeline-Aufgaben aufgewendet werden, wie z. B. die Orchestrierung von Pipelines, den Aufbau der Infrastruktur und die Schaffung der dahinter stehenden Infrastruktur. DevOps Sie können diese arbeitsintensiven Aufgaben auslagern, AWS DDK sodass Sie sich auf das Schreiben von Code und andere wichtige Aktivitäten konzentrieren können.

Voraussetzungen und Einschränkungen

Voraussetzungen

Versionen der Produkte

Architektur

Technologie-Stack

Zielarchitektur

Das folgende Diagramm zeigt den ereignisgesteuerten Prozess, der Google Analytics-Daten aufnimmt, transformiert und analysiert.

Das Diagramm zeigt den folgenden Workflow:

Tools

AWSWerkzeuge

Code

Der Code für dieses Muster ist in den GitHub AWS DataOps Repositorys Development Kit (DDK) und Analysieren von Google Analytics-Daten mit Amazon AppFlow, Amazon Athena und AWS DataOps Development Kit verfügbar.

Epen

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

Fehlerbehebung

Zugehörige Ressourcen

Zusätzliche Informationen

AWSDDKDaten-Pipelines bestehen aus einer oder mehreren Stufen. In den folgenden Codebeispielen verwenden Sie, AppFlowIngestionStage um Daten aus Google Analytics aufzunehmen, SqsToLambdaStage die Datentransformation durchzuführen und AthenaSQLStage die Athena-Abfrage auszuführen.

Zunächst werden die Phasen Datentransformation und Datenaufnahme erstellt, wie das folgende Codebeispiel zeigt:

        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)
            ],
        )

Als Nächstes wird das DataPipeline Konstrukt verwendet, um die Stufen mithilfe von EventBridge Regeln miteinander zu verbinden, wie das folgende Codebeispiel zeigt:

        (
            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)
        )

Weitere Codebeispiele finden Sie im Repository GitHub Analysieren von Google Analytics-Daten mit Amazon AppFlow, Amazon Athena und AWS DataOps Development Kit.