Crie um pipeline de dados para ingerir, transformar e analisar dados do Google Analytics usando o AWS DataOps Development Kit

Criado por Anton Kukushkin (AWS) e Rudy Puig () AWS

Resumo

Esse padrão descreve como criar um pipeline de dados para ingerir, transformar e analisar dados do Google Analytics usando o AWS DataOps Development Kit (DDK) e outros AWS serviços. AWSDDKÉ uma estrutura de desenvolvimento de código aberto que ajuda você a criar fluxos de trabalho de dados e uma arquitetura de dados moderna em. AWS Um dos principais objetivos do AWS DDK é economizar o tempo e o esforço que normalmente são dedicados às tarefas trabalhosas do pipeline de dados, como orquestrar pipelines, construir infraestrutura e criar a base dessa infraestrutura. DevOps Você pode transferir essas tarefas trabalhosas para AWS DDK se concentrar em escrever código e outras atividades de alto valor.

Pré-requisitos e limitações

Pré-requisitos

Versões do produto

Arquitetura

Pilha de tecnologia

Arquitetura de destino

O diagrama a seguir mostra o processo orientado por eventos que ingere, transforma e analisa os dados do Google Analytics.

O diagrama mostra o seguinte fluxo de trabalho:

Ferramentas

AWSferramentas

Código

O código desse padrão está disponível nos repositórios GitHub AWS DataOps Development Kit (DDK) e Analisando dados do Google Analytics com Amazon AppFlow, Amazon Athena e AWS DataOps Development Kit.

Épicos

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

Solução de problemas

Recursos relacionados

Mais informações

AWSDDKos pipelines de dados são compostos por um ou vários estágios. Nos exemplos de código a seguir, você usa AppFlowIngestionStage para ingerir dados do Google Analytics, SqsToLambdaStage lidar com a transformação de dados e AthenaSQLStage para executar a consulta do Athena.

Primeiro, os estágios de transformação e ingestão de dados são criados, conforme mostra o exemplo de código a seguir:

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

Em seguida, a DataPipeline construção é usada para “conectar” os estágios usando EventBridge regras, como mostra o exemplo de código a seguir:

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

Para ver mais exemplos de código, consulte o repositório GitHub Analisando dados do Google Analytics com Amazon AppFlow, Amazon Athena e AWS DataOps Development Kit.