Sources                     Ingestion              Storage           Transform          Serve
─────────                   ─────────              ───────           ─────────          ─────
RDS (PostgreSQL) ──▶ Glue ETL Job ──▶ S3 (bronze) ──▶ Glue ETL / dbt ──▶ S3 (gold) ──▶ Athena
Kinesis Streams ──▶ Firehose ──────▶ S3 (bronze)                               └──▶ Redshift
Third-party APIs ──▶ Lambda ───────▶ S3 (bronze)                               └──▶ QuickSight

                    Glue Data Catalog (schema registry, table metadata)
                    Lake Formation (access control, permissions)

import sys
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.transforms import *
from pyspark.context import SparkContext

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(sys.argv[0], {"JOB_NAME": "orders_bronze_to_silver"})

# Read from Glue Data Catalog
orders_dyf = glueContext.create_dynamic_frame.from_catalog(
    database="raw_db",
    table_name="raw_orders"
)

# Transform
orders_df = orders_dyf.toDF()
silver_df = (
    orders_df
    .filter(orders_df.order_id.isNotNull())
    .withColumn("amount", orders_df.amount.cast("decimal(10,2)"))
    .withColumn("order_date", orders_df.order_date.cast("date"))
    .dropDuplicates(["order_id"])
)

# Write to S3 as Parquet, partitioned
silver_df.write \
    .mode("overwrite") \
    .partitionBy("year", "month") \
    .parquet("s3://data-lake/silver/orders/")

job.commit()

-- Query S3 data via Glue Data Catalog — no infrastructure to manage
SELECT
    order_date,
    COUNT(*) AS order_count,
    SUM(amount) AS total_revenue
FROM silver_db.orders
WHERE year = '2025' AND month = '06'
GROUP BY order_date
ORDER BY order_date;

-- Grant fine-grained access (column-level and row-level)
Grant SELECT on database silver_db
    table orders
    columns (order_id, order_date, amount, status)
    to IAM role data-analyst-role
    with data filter (country = 'GB')

Cloud SQL / APIs ──▶ Dataflow ──▶ GCS (bronze) ──▶ Dataform / dbt ──▶ BigQuery (gold)
Pub/Sub ──────────▶ Dataflow ──▶ BigQuery (streaming insert)
                                                                └──▶ Looker (BI)

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

options = PipelineOptions([
    "--runner=DataflowRunner",
    "--project=my-gcp-project",
    "--region=europe-west1",
    "--temp_location=gs://my-bucket/temp/",
])

with beam.Pipeline(options=options) as p:
    (
        p
        | "Read from Pub/Sub" >> beam.io.ReadFromPubSub(topic="projects/my-project/topics/orders")
        | "Parse JSON" >> beam.Map(lambda msg: json.loads(msg))
        | "Filter valid" >> beam.Filter(lambda order: order.get("amount", 0) > 0)
        | "Window" >> beam.WindowInto(beam.window.FixedWindows(300))  # 5-minute windows
        | "Count per status" >> beam.combiners.Count.PerKey(lambda order: order["status"])
        | "Write to BigQuery" >> beam.io.WriteToBigQuery(
            table="my_project:analytics.order_counts",
            schema="status:STRING,count:INTEGER",
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
        )
    )

SQL Server / APIs ──▶ Data Factory ──▶ ADLS Gen2 (bronze) ──▶ Synapse Spark / dbt ──▶ Synapse SQL pool (gold)
Event Hubs ─────────▶ Stream Analytics ──▶ ADLS Gen2 / Synapse                       └──▶ Power BI
                                                                                       
                     Purview (catalog, lineage, governance)