Crea una pipeline di dati per importare, trasformare e analizzare i dati di Google Analytics utilizzando il AWS DataOps Development Kit

Creato da Anton Kukushkin (AWS) e Rudy Puig () AWS

Riepilogo

Questo modello descrive come creare una pipeline di dati per importare, trasformare e analizzare i dati di Google Analytics utilizzando il AWS DataOps Development Kit () DDK e altri servizi. AWS AWSDDKÈ un framework di sviluppo open source che ti aiuta a creare flussi di lavoro di dati e un'architettura di dati moderna su. AWS Uno degli obiettivi principali di AWS DDK è quello di farti risparmiare tempo e fatica tipicamente dedicati alle attività di pipeline di dati ad alta intensità di manodopera, come l'orchestrazione delle pipeline, la creazione di infrastrutture e la creazione dell'infrastruttura alla base di tale infrastruttura. DevOps Puoi delegare queste attività ad alta intensità di lavoro in AWS DDK modo da poterti concentrare sulla scrittura di codice e su altre attività di alto valore.

Prerequisiti e limitazioni

Prerequisiti

Versioni del prodotto

Architettura

stack tecnologico

Architettura Target

Il diagramma seguente mostra il processo basato sugli eventi che acquisisce, trasforma e analizza i dati di Google Analytics.

Il diagramma mostra il flusso di lavoro seguente:

Strumenti

AWSstrumenti

Codice

Il codice per questo pattern è disponibile negli archivi GitHub AWS DataOps Development Kit (DDK) e Analyzing Google Analytics con Amazon AppFlow, Amazon Athena AWS DataOps e Development Kit.

Epiche

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

Risoluzione dei problemi

Risorse correlate

Informazioni aggiuntive

AWSDDKle pipeline di dati sono composte da una o più fasi. Nei seguenti esempi di codice, li usi AppFlowIngestionStage per importare dati da Google Analytics, SqsToLambdaStage gestire la trasformazione dei dati ed AthenaSQLStage eseguire la query Athena.

Innanzitutto, vengono create le fasi di trasformazione e ingestione dei dati, come mostra il seguente esempio di codice:

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

Successivamente, il DataPipeline costrutto viene utilizzato per «collegare» gli stadi utilizzando EventBridge delle regole, come mostra il seguente esempio di codice:

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

Per altri esempi di codice, consulta il GitHub repository Analisi dei dati di Google Analytics con Amazon AppFlow, Amazon Athena AWS DataOps e Development Kit.