Hands-On Project: Build a Data Pipeline¶

# ingest_api.py
import requests
import json
import boto3  # or google.cloud.storage, azure.storage.blob
from datetime import datetime

def ingest_from_api():
    """Fetch data from a REST API and store as JSON in cloud storage."""
    url = "https://api.example.com/v1/data"
    headers = {"Authorization": "Bearer YOUR_TOKEN"}

    response = requests.get(url, headers=headers)
    response.raise_for_status()
    data = response.json()

    # Add metadata
    payload = {
        "ingested_at": datetime.utcnow().isoformat(),
        "source": "api.example.com",
        "record_count": len(data["results"]),
        "data": data["results"]
    }

    # Store in S3 (raw zone)
    s3 = boto3.client("s3")
    key = f"raw/api-data/{datetime.utcnow().strftime('%Y/%m/%d/%H%M%S')}.json"
    s3.put_object(
        Bucket="my-data-lake",
        Key=key,
        Body=json.dumps(payload),
        ContentType="application/json"
    )
    return key

# glue_transform.py
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.context import SparkContext
from pyspark.sql import functions as F

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
job.init(args["JOB_NAME"], args)

# Read raw data
raw_df = spark.read.json("s3://my-data-lake/raw/api-data/2024/01/15/")

# Transform
transformed_df = (
    raw_df
    .select(F.explode("data").alias("record"))
    .select(
        F.col("record.id").alias("record_id"),
        F.col("record.name").alias("name"),
        F.col("record.amount").cast("decimal(10,2)").alias("amount"),
        F.to_timestamp("record.created_at").alias("created_at"),
        F.current_timestamp().alias("processed_at")
    )
    .filter(F.col("amount").isNotNull())
    .dropDuplicates(["record_id"])
)

# Write to staging
transformed_df.write \
    .mode("overwrite") \
    .partitionBy("created_at") \
    .parquet("s3://my-data-lake/staging/records/")

job.commit()

# dataflow_transform.py
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

class ParseRecord(beam.DoFn):
    def process(self, element):
        import json
        record = json.loads(element)
        if record.get("amount") is not None:
            yield {
                "record_id": record["id"],
                "name": record["name"],
                "amount": float(record["amount"]),
                "created_at": record["created_at"]
            }

def run():
    options = PipelineOptions([
        "--runner=DataflowRunner",
        "--project=my-project",
        "--region=us-central1",
        "--temp_location=gs://my-data-lake/temp/"
    ])

    with beam.Pipeline(options=options) as p:
        (
            p
            | "ReadFromGCS" >> beam.io.ReadFromText("gs://my-data-lake/raw/api-data/*.json")
            | "ParseRecords" >> beam.ParDo(ParseRecord())
            | "WriteToBigQuery" >> beam.io.WriteToBigQuery(
                "my-project:my_dataset.records",
                schema="record_id:STRING,name:STRING,amount:FLOAT,created_at:TIMESTAMP",
                write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
            )
        )

if __name__ == "__main__":
    run()

# data_quality.py
import great_expectations as gx

context = gx.get_context()

# Define expectations
validator = context.sources.pandas_default.read_parquet(
    "s3://my-data-lake/staging/records/"
)

validator.expect_column_values_to_not_be_null("record_id")
validator.expect_column_values_to_be_unique("record_id")
validator.expect_column_values_to_be_between("amount", min_value=0, max_value=1000000)
validator.expect_column_values_to_match_regex("name", r"^[A-Za-z\s]+$")

results = validator.validate()

if not results.success:
    raise Exception(f"Data quality check failed: {results}")

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

dag = DAG(
    "data_pipeline",
    schedule_interval="0 6 * * *",  # Daily at 6 AM
    start_date=datetime(2024, 1, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)}
)

ingest = PythonOperator(task_id="ingest", python_callable=ingest_from_api, dag=dag)
transform = PythonOperator(task_id="transform", python_callable=run_transform, dag=dag)
quality = PythonOperator(task_id="quality_check", python_callable=run_quality_checks, dag=dag)
load = PythonOperator(task_id="load", python_callable=load_to_warehouse, dag=dag)

ingest >> transform >> quality >> load