Membangun pipeline data untuk menyerap, mengubah, dan menganalisis data Google Analytics menggunakan AWS DataOps Development Kit

Anton Kukushkin dan Rudy Puig, Amazon Web Services

Ringkasan

Pola ini menjelaskan cara membuat pipeline data untuk menyerap, mengubah, dan menganalisis data Google Analytics dengan menggunakan AWS DataOps Development Kit (AWS DDK) dan lainnya. Layanan AWS AWS DDK adalah kerangka kerja pengembangan sumber terbuka yang membantu Anda membangun alur kerja data dan arsitektur data modern. AWS Salah satu tujuan utama AWS DDK adalah untuk menghemat waktu dan upaya yang biasanya dikhususkan untuk tugas-tugas pipa data padat karya, seperti mengatur jaringan pipa, membangun infrastruktur, dan menciptakan infrastruktur di balik itu. DevOps Anda dapat menurunkan tugas-tugas padat karya ini ke AWS DDK sehingga Anda dapat fokus pada penulisan kode dan aktivitas bernilai tinggi lainnya.

Prasyarat dan batasan

Prasyarat

Versi produk

Arsitektur

Tumpukan teknologi

Arsitektur target

Diagram berikut menunjukkan proses berbasis peristiwa yang menyerap, mengubah, dan menganalisis data Google Analytics.

Diagram menunjukkan alur kerja berikut:

Alat

AWS alat-alat

Kode

Kode untuk pola ini tersedia di GitHub AWS DataOps Development Kit (AWS DDK) dan Menganalisis data Google Analytics dengan repositori Amazon AppFlow, Amazon Athena, AWS DataOps dan Development Kit.

Epik

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

Pemecahan Masalah

Sumber daya terkait

Informasi tambahan

AWS Pipa data DDK terdiri dari satu atau banyak tahapan. Dalam contoh kode berikut, Anda gunakan AppFlowIngestionStage untuk menyerap data dari Google Analytics, SqsToLambdaStage menangani transformasi data, dan AthenaSQLStage menjalankan kueri Athena.

Pertama, transformasi data dan tahap konsumsi dibuat, seperti contoh kode berikut menunjukkan:

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

Selanjutnya, DataPipeline konstruksi digunakan untuk “menghubungkan” tahapan bersama-sama dengan menggunakan EventBridge aturan, seperti contoh kode berikut menunjukkan:

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

Untuk contoh kode lainnya, lihat GitHub Menganalisis data Google Analytics dengan Amazon AppFlow, Amazon Athena, dan repositori AWS DataOps Development Kit.