Big Data & Analytics

import os

import aws_cdk as cdk
from aws_cdk import (
    aws_events as events,
    aws_events_targets as targets,
    aws_glue as glue,
    aws_iam as iam,
    aws_lambda as lambda_,
    aws_s3 as s3,
    aws_s3_assets as s3_assets,
)
from constructs import Construct

GLUE_JOBS_DIR = os.path.join(os.path.dirname(__file__), "..", "glue_jobs")
LAMBDAS_DIR = os.path.join(os.path.dirname(__file__), "..", "lambdas")


class DataLakeStack(cdk.Stack):
    def __init__(self, scope: Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        # Single bucket, two zones by prefix: raw/ (JSON, as-landed) and
        # curated/ (Parquet, partitioned). DESTROY + auto_delete_objects means
        # `cdk destroy` leaves nothing behind to clean up by hand.
        bucket = s3.Bucket(
            self, "DataLakeBucket",
            removal_policy=cdk.RemovalPolicy.DESTROY,
            auto_delete_objects=True,
            # Lets EventBridge see this bucket's "Object Created" events.
            event_bridge_enabled=True,
        )

        # Shared metadata: one database, two tables (raw_orders,
        # curated_orders), the same "define it once" idea the lesson teaches.
        database = glue.CfnDatabase(
            self, "AnalyticsDatabase",
            catalog_id=self.account,
            database_input=glue.CfnDatabase.DatabaseInputProperty(
                name="ecommerce_analytics",
                description="Raw and curated order tables for the data lake showcase",
            ),
        )

        # Crawlers: both assume the same minimal role, scoped to read the
        # bucket and manage tables in this database.
        crawler_role = iam.Role(self, "CrawlerRole",
            assumed_by=iam.ServicePrincipal("glue.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AWSGlueServiceRole"),
            ],
        )

        bucket.grant_read(crawler_role)

        raw_crawler = glue.CfnCrawler(self, "RawOrdersCrawler",
            name="raw-orders-crawler",
            role=crawler_role.role_arn,
            database_name=database.ref,
            targets=glue.CfnCrawler.TargetsProperty(
                s3_targets=[glue.CfnCrawler.S3TargetProperty(
                    path=f"s3://{bucket.bucket_name}/raw/orders/",
                )],
            ),
            # Crawler names the table after the last path segment ("orders"),
            # so table_prefix="raw_" yields "raw_orders"
            table_prefix="raw_",
            schema_change_policy=glue.CfnCrawler.SchemaChangePolicyProperty(
                update_behavior="UPDATE_IN_DATABASE",
                delete_behavior="LOG",
            ),
        )

        raw_crawler.node.add_dependency(bucket)

        curated_crawler = glue.CfnCrawler(self, "CuratedOrdersCrawler",
            name="curated-orders-crawler",
            role=crawler_role.role_arn,
            database_name=database.ref,
            targets=glue.CfnCrawler.TargetsProperty(
                s3_targets=[glue.CfnCrawler.S3TargetProperty(
                    path=f"s3://{bucket.bucket_name}/curated/orders/",
                )],
            ),
            table_prefix="curated_",
            schema_change_policy=glue.CfnCrawler.SchemaChangePolicyProperty(
                update_behavior="UPDATE_IN_DATABASE",
                delete_behavior="LOG",
            ),
        )

        curated_crawler.node.add_dependency(bucket)

        # Near-real-time pipeline: rather than running crawlers/jobs on a
        # schedule (polling) or by hand, EventBridge rules fire the moment a
        # new object lands in `raw/orders/` or `curated/orders/`, and a small
        # Lambda starts the matching crawler or job, every stage typically
        # refreshes within minutes of new data arriving, at effectively no
        # extra cost. See PipelineTriggerFn's targets below for the full chain.
        pipeline_trigger_fn = lambda_.Function(self, "PipelineTriggerFn",
            runtime=lambda_.Runtime.PYTHON_3_12,
            handler="handler.handler",
            code=lambda_.Code.from_asset(os.path.join(LAMBDAS_DIR, "pipeline_trigger")),
            timeout=cdk.Duration.seconds(30),
        )

        pipeline_trigger_fn.add_to_role_policy(
            iam.PolicyStatement(
                actions=["glue:StartCrawler"],
                resources=[
                    self.format_arn(service="glue", resource="crawler", resource_name=raw_crawler.ref),
                    self.format_arn(service="glue", resource="crawler", resource_name=curated_crawler.ref),
                ],
            )
        )

        # New raw JSON triggers two things in parallel: catalog it (raw_crawler)
        # and curate it (curator_job, added once it exists further below)
        raw_landing_rule = events.Rule(self, "RawOrderLandingRule",
            event_pattern=events.EventPattern(
                source=["aws.s3"],
                detail_type=["Object Created"],
                detail={
                    "bucket": {"name": [bucket.bucket_name]},
                    "object": {"key": [{"prefix": "raw/orders/"}]},
                },
            ),
        )

        raw_landing_rule.add_target(
            targets.LambdaFunction(pipeline_trigger_fn,
                event=events.RuleTargetInput.from_object({"crawlerName": raw_crawler.ref}),
            )
        )

        curated_landing_rule = events.Rule(self, "CuratedOrderLandingRule",
            event_pattern=events.EventPattern(
                source=["aws.s3"],
                detail_type=["Object Created"],
                detail={
                    "bucket": {"name": [bucket.bucket_name]},
                    "object": {"key": [{"prefix": "curated/orders/"}]},
                },
            ),
        )

        curated_landing_rule.add_target(
            targets.LambdaFunction(pipeline_trigger_fn,
                event=events.RuleTargetInput.from_object({"crawlerName": curated_crawler.ref}),
            )
        )

        # ETL: a Python Shell job (NOT Spark/EMR). At this data scale a
        # Spark job would spend most of its billed minute on cluster startup;
        # Python Shell runs at a fraction of a DPU and finishes in seconds,
        # the cheapest "real" Glue ETL option, and the right one here.
        job_role = iam.Role(self, "JobRole",
            assumed_by=iam.ServicePrincipal("glue.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AWSGlueServiceRole"),
            ],
        )

        bucket.grant_read_write(job_role)

        job_script_asset = s3_assets.Asset(self, "CurateOrdersScriptAsset",
            path=os.path.join(GLUE_JOBS_DIR, "curate_orders.py"),
        )

        job_script_asset.grant_read(job_role)

        curator_job = glue.CfnJob(self, "OrdersCuratorJob",
            name="orders-curator-job",
            role=job_role.role_arn,
            command=glue.CfnJob.JobCommandProperty(
                name="pythonshell",
                python_version="3.9",
                script_location=job_script_asset.s3_object_url,
            ),
            default_arguments={
                "--TempDir": f"s3://{bucket.bucket_name}/glue-temp/",
                "--job-bookmark-option": "job-bookmark-disable",
                "--library-set": "analytics",
                "--BUCKET_NAME": bucket.bucket_name,
                "--DATABASE_NAME": database.ref,
            },
            glue_version="3.0",
            max_capacity=0.0625,
            max_retries=0,
            timeout=10,
        )

        curator_job.node.add_dependency(job_script_asset)

        # Completes the raw-landing chain: the same new-JSON event that starts
        # raw_crawler also starts curator_job, so curation runs in parallel
        # with cataloging rather than waiting on it (curate_orders.py reads
        # straight from S3, it has no dependency on the raw_orders table).
        pipeline_trigger_fn.add_to_role_policy(
            iam.PolicyStatement(
                actions=["glue:StartJobRun"],
                resources=[self.format_arn(service="glue", resource="job", resource_name=curator_job.ref)],
            )
        )

        raw_landing_rule.add_target(
            targets.LambdaFunction(pipeline_trigger_fn,
                event=events.RuleTargetInput.from_object({"jobName": curator_job.ref}),
            )
        )

        cdk.CfnOutput(self, "BucketName",
            value=bucket.bucket_name,
            description="Data lake bucket, pass to generate_orders.py",
        )

        cdk.CfnOutput(self, "DatabaseName",
            value=database.ref,
            description="Glue Data Catalog database, pass to compare_queries.py",
        )

        cdk.CfnOutput(self, "RawOrdersCrawlerName",
            value=raw_crawler.ref,
            description="Crawler that catalogs the raw/orders/ JSON zone into raw_orders",
        )

        cdk.CfnOutput(self, "CuratedOrdersCrawlerName",
            value=curated_crawler.ref,
            description="Crawler that catalogs the curated/orders/ Parquet zone into curated_orders",
        )

        cdk.CfnOutput(self, "PipelineTriggerFunctionName",
            value=pipeline_trigger_fn.function_name,
            description="Lambda that starts the matching crawler/job the moment a new object lands in its watched prefix",
        )

        cdk.CfnOutput(self, "OrdersCuratorJobName",
            value=curator_job.ref,
            description="Python Shell job that converts raw JSON orders into partitioned, curated Parquet",
        )

-- Partition projection: Athena computes partition locations from the S3
-- key pattern itself, no crawler run (or "MSCK REPAIR TABLE") needed after
-- every new day of data lands
CREATE EXTERNAL TABLE analytics_db.orders (
    order_id      string,
    customer_id   string,
    total_amount  double,
    status        string
)
PARTITIONED BY (year string, month string, day string)
STORED AS PARQUET
LOCATION 's3://my-curated-bucket/curated/orders/'
TBLPROPERTIES (
    'projection.enabled'        = 'true',
    'projection.year.type'      = 'integer',
    'projection.year.range'     = '2024,2030',
    'projection.month.type'     = 'integer',
    'projection.month.range'    = '1,12',
    'projection.month.digits'   = '2',
    'projection.day.type'       = 'integer',
    'projection.day.range'      = '1,31',
    'projection.day.digits'     = '2'
);

-- Cost-aware query: the WHERE clause prunes partitions, so Athena scans
-- only the matching day's Parquet files rather than the whole table
SELECT
    customer_id,
    COUNT(*)          AS order_count,
    SUM(total_amount) AS daily_total
FROM analytics_db.orders
WHERE year = 2026 AND month = 6 AND day = 5
  AND status = 'COMPLETED'
GROUP BY customer_id
ORDER BY daily_total DESC
LIMIT 10;

-- The IAM role attached to the Redshift cluster needs glue:GetTable,
-- glue:GetDatabase, and s3:GetObject on the curated bucket (scope this
-- down from AWSGlueConsoleFullAccess with a custom policy in production)

-- Register the Glue Data Catalog database as an external schema: Redshift
-- now "sees" the same tables Athena queries, with zero data movement
CREATE EXTERNAL SCHEMA spectrum_orders
FROM DATA CATALOG
DATABASE 'analytics_db'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftSpectrumRole'
REGION 'us-east-1';

-- Federated query: joins a native Redshift table (loaded, distributed,
-- fast) with an external Spectrum table (S3-backed, no load step) in
-- one statement, blending warehouse and data-lake data transparently
SELECT
    c.segment,
    DATE_TRUNC('month', o.order_date) AS month,
    SUM(o.total_amount)               AS revenue
FROM customers c
JOIN spectrum_orders.orders o
    ON o.customer_id = c.customer_id
WHERE o.year = 2026
GROUP BY c.segment, DATE_TRUNC('month', o.order_date)
ORDER BY month, revenue DESC;