Créez un pipeline de données pour ingérer, transformer et analyser les données de Google Analytics à l'aide du kit de AWS DataOps développement

Créé par Anton Kukushkin (AWS) et Rudy Puig () AWS

Récapitulatif

Ce modèle décrit comment créer un pipeline de données pour ingérer, transformer et analyser les données Google Analytics à l'aide du kit de AWS DataOps développement (DDK) et d'autres AWS services. AWSDDKIl s'agit d'un framework de développement open source qui vous aide à créer des flux de travail de données et une architecture de données moderne. AWS L'un des principaux objectifs du AWS DDK est de vous faire économiser le temps et les efforts généralement consacrés aux tâches de pipeline de données à forte intensité de main-d'œuvre, telles que l'orchestration de pipelines, la création d'infrastructures et la création de l'infrastructure DevOps sous-jacente. Vous pouvez vous décharger de ces tâches fastidieuses AWS DDK afin de vous concentrer sur l'écriture de code et d'autres activités à forte valeur ajoutée.

Conditions préalables et limitations

Prérequis

Versions du produit

Architecture

Pile technologique

Architecture cible

Le schéma suivant montre le processus piloté par les événements qui ingère, transforme et analyse les données de Google Analytics.

Le schéma suivant illustre le flux de travail suivant :

Outils

AWSoutils

Code

Le code de ce modèle est disponible dans les GitHub AWS DataOps référentiels Development Kit (DDK) et Analyzing Google Analytics with Amazon AppFlow, Amazon Athena et AWS DataOps Development Kit.

Épopées

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

Résolution des problèmes

Ressources connexes

Informations supplémentaires

AWSDDKles pipelines de données sont composés d'une ou de plusieurs étapes. Dans les exemples de code suivants, vous les utilisez AppFlowIngestionStage pour ingérer des données provenant de Google Analytics, SqsToLambdaStage pour gérer la transformation des données et AthenaSQLStage pour exécuter la requête Athena.

Tout d'abord, les étapes de transformation et d'ingestion des données sont créées, comme le montre l'exemple de code suivant :

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

Ensuite, la DataPipeline construction est utilisée pour « relier » les étapes entre elles en utilisant des EventBridge règles, comme le montre l'exemple de code suivant :

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

Pour d'autres exemples de code, consultez le GitHub référentiel Analyse des données Google Analytics avec Amazon AppFlow, Amazon Athena et AWS DataOps le référentiel Development Kit.