Comment orchestrer un ETL avancé avec AWS Glue en 2026

mkdir glue-etl-pipeline && cd glue-etl-pipeline
pip install aws-cdk-lib constructs
cdk init app --language python
source .venv/bin/activate
pip install -r requirements.txt
cdk bootstrap aws://$(aws sts get-caller-identity --query Account --output text)/$(aws ec2 describe-regions --query 'Regions[0].RegionName' --output text)

from aws_cdk import Stack, aws_s3 as s3, aws_iam as iam, aws_glue as glue, aws_logs as logs, CfnOutput

from aws_cdk import App, Stack
from constructs import Construct

class GluePipelineStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Bucket S3 pour input/output
        self.input_bucket = s3.Bucket(self, "GlueInputBucket",
            bucket_name=f"glue-demo-input-{self.account}-{self.region}",
            block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
            versioned=True,
            removal_policy=RemovalPolicy.DESTROY
        )
        self.output_bucket = s3.Bucket(self, "GlueOutputBucket",
            bucket_name=f"glue-demo-output-{self.account}-{self.region}",
            block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
            removal_policy=RemovalPolicy.DESTROY
        )
        CfnOutput(self, "InputBucketName", value=self.input_bucket.bucket_name)
        CfnOutput(self, "OutputBucketName", value=self.output_bucket.bucket_name)

        # IAM Role pour Glue
        self.glue_role = iam.Role(self, "GlueETLRole",
            assumed_by=iam.ServicePrincipal("glue.amazonaws.com"),
            managed_policies=[
                iam.ServicePrincipal("glue.amazonaws.com").add_managed_policy(iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AWSGlueServiceRole")),
            ]
        )
        self.glue_role.add_to_policy(iam.PolicyStatement(
            actions=["s3:GetObject", "s3:PutObject", "s3:DeleteObject"],
            resources=[self.input_bucket.bucket_arn + "/*", self.output_bucket.bucket_arn + "/*"]
        ))
        self.glue_role.add_to_policy(iam.PolicyStatement(
            actions=["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
            resources=["*"]
        ))
        CfnOutput(self, "GlueRoleArn", value=self.glue_role.role_arn)

app = App()
GluePipelineStack(app, "GluePipelineStack")
app.synth()

#!/bin/bash
BUCKET_INPUT=$(aws cloudformation describe-stacks --stack-name GluePipelineStack --query 'Stacks[0].Outputs[?OutputKey==`InputBucketName`].OutputValue' --output text)
aws s3 cp nyc_taxi_sample.csv s3://$BUCKET_INPUT/input/ --recursive
cat > glue_etl_job.py << 'EOF'
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Lire CSV brut
ds = glueContext.create_dynamic_frame.from_catalog(database="glue_demo_db", table_name="nyc_taxi_raw")

# Transformer : Convert to DF, clean nulls/duplicates, aggregate
spark_df = ds.toDF()
spark_df_clean = spark_df.dropDuplicates(["pickup_datetime", "pickup_longitude"]).na.drop(subset=["total_amount"])
spark_df_agg = spark_df_clean.groupBy("pickup_datetime").agg({"total_amount": "sum", "passenger_count": "avg"}).withColumnRenamed("sum(total_amount)", "total_revenue").withColumnRenamed("avg(passenger_count)", "avg_passengers")

# Écrire Parquet partitionné
dynamic_frame_agg = DynamicFrame.fromDF(spark_df_agg.coalesce(1), glueContext, "agg_df")
glueContext.write_dynamic_frame.from_options(dynamic_frame_agg,
    connection_type="s3",
    connection_options={"path": "s3://OUTPUT_BUCKET/output/parquet/"},
    format="parquet")

job.commit()
EOF
aws s3 cp glue_etl_job.py s3://$BUCKET_INPUT/scripts/

from aws_cdk import Stack, aws_glue as glue, CfnOutput
from aws_cdk import RemovalPolicy

class CrawlerStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, input_bucket_name: str, glue_role_arn: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Database Glue
        database = glue.CfnDatabase(self, "GlueDemoDB",
            catalog_id=self.account,
            database_input=glue.CfnDatabase.DatabaseInputProperty(
                name="glue_demo_db",
                description="DB pour NYC Taxi ETL"
            )
        )

        # Crawler
        crawler = glue.CfnCrawler(self, "NYCTaxiCrawler",
            role=glue_role_arn,
            database_name="glue_demo_db",
            description="Crawler pour input S3 NYC Taxi",
            targets=glue.CfnCrawler.TargetsProperty(
                s3_targets=[glue.CfnCrawler.S3TargetProperty(
                    path=f"s3://{input_bucket_name}/input/"
                )]
            ),
            schedule=glue.CfnCrawler.ScheduleProperty(schedule_expression="rate(1 day)"),
            configuration="{\"Version\":1.0,\"CrawlerOutput\":{\"Partitions\":{\"AddOrUpdateBehavior\":\"InheritFromTable\"},\"Tables\":{\"AddOrUpdateBehavior\":\"MergeNewColumns\"}}}",
            recrawl_policy=glue.CfnCrawler.RecrawlPolicyProperty(
                recrawl_behavior="CRAWL_EVENT_MODE"
            )
        )
        crawler.add_depends_on(database)
        CfnOutput(self, "CrawlerName", value=crawler.ref)

# Intégrer dans app
app = App()
# Assume props from previous stack via SSM or manual
CrawlerStack(app, "CrawlerStack", input_bucket_name="glue-demo-input-XXXX-us-east-1", glue_role_arn="arn:aws:iam::XXXX:role/GlueETLRole")
app.synth()

from aws_cdk import Stack, aws_glue as glue, aws_iam as iam, CfnOutput

class JobStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, input_bucket_name: str, output_bucket_name: str, glue_role_arn: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        job = glue.CfnJob(self, "NYCTaxiETLJob",
            role_arn=glue_role_arn,
            command=glue.CfnJob.CommandProperty(
                name="glueetl",
                script_location=f"s3://{input_bucket_name}/scripts/glue_etl_job.py",
                python_version="3"
            ),
            default_arguments={
                "--TempDir": f"s3://{input_bucket_name}/temp/",
                "--OUTPUT_BUCKET": output_bucket_name,
                "--job-bookmark-option": "job-bookmark-enable",
                "--enable-job-insights": "true",
                "--datalake-defaults": "true"
            },
            glue_version="4.0",  # PySpark 3.3+ en 2026
            number_of_workers=10,
            worker_type="G.2X",
            max_capacity=10.0,
            timeout=60,
            max_retries=3
        )
        CfnOutput(self, "JobName", value=job.ref)

app = App()
JobStack(app, "JobStack", "glue-demo-input-XXXX-us-east-1", "glue-demo-output-XXXX-us-east-1", "arn:aws:iam::XXXX:role/GlueETLRole")
app.synth()

from aws_cdk import Stack, aws_glue as glue, CfnOutput

class WorkflowStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, crawler_name: str, job_name: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        workflow = glue.CfnWorkflow(self, "NYCTaxiWorkflow",
            description="Workflow ETL NYC Taxi: Crawler -> Job",
            default_run_properties={"conf": "spark.sql.adaptive.enabled=true"}
        )

        trigger = glue.CfnTrigger(self, "CrawlerSuccessTrigger",
            name="trigger-crawler-to-job",
            workflow_name=workflow.ref,
            type="CONDITIONAL",
            start_on=glue.CfnTrigger.StartOnProperty(
                crawler_names=[crawler_name]
            ),
            actions=[glue.CfnTrigger.ActionProperty(job_name=job_name)],
            predicate=glue.CfnTrigger.PredicateProperty(
                conditions=[glue.CfnTrigger.ConditionProperty(
                    logical_operator="EQUALS",
                    job_name="",
                    state="SUCCEEDED",
                    crawler_name=crawler_name
                )]
            )
        )
        CfnOutput(self, "WorkflowName", value=workflow.ref)

app = App()
WorkflowStack(app, "WorkflowStack", "NYCTaxiCrawler", "NYCTaxiETLJob")
app.synth()

cd glue_pipeline
cdk deploy GluePipelineStack CrawlerStack JobStack WorkflowStack --require-approval never
aws glue start-workflow-run --name $(aws cloudformation describe-stacks --stack-name WorkflowStack --query 'Stacks[0].Outputs[?OutputKey==`WorkflowName`].OutputValue' --output text)
aws glue get-workflow-run --name WORKFLOW_NAME --run-id $(aws glue start-workflow-run --name WORKFLOW_NAME --query 'RunId' --output text) --query 'Run.Parts[0].ExecutionStatus'