Cloud Data Platforms

← Previous: SQL for Data Engineering | Next: Streaming & Real-Time Data →

Introduction

Cloud platforms have transformed data engineering. Instead of managing on-premise hardware and databases, we now leverage fully managed services that scale automatically. In my work, I've used AWS, Azure, GCP, Snowflake, and Databricks extensively. Each has strengths and trade-offs.

This article covers the core cloud data services I use regularly, with practical Python 3.12 examples. I'll share real experiences, not theoretical comparisons—what actually works in production environments.

AWS Data Services

AWS is my most-used cloud platform. Here are the essential data engineering services:

Amazon S3 - Object Storage

S3 is the foundation of most AWS data architectures. I use it for:

• Data lake storage (raw, processed, curated zones)

• Pipeline staging (temporary file processing)

• Backup and archival (long-term retention)

# s3_operations.py
import boto3
from botocore.exceptions import ClientError
import pandas as pd
from io import StringIO, BytesIO
from typing import List, Optional
from pathlib import Path

class S3Manager:
    """Manage S3 operations for data engineering"""
    
    def __init__(self, bucket_name: str, region: str = 'us-east-1'):
        """
        Initialize S3 manager.
        
        Args:
            bucket_name: S3 bucket name
            region: AWS region
        """
        self.bucket_name = bucket_name
        self.s3_client = boto3.client('s3', region_name=region)
        self.s3_resource = boto3.resource('s3', region_name=region)
        self.bucket = self.s3_resource.Bucket(bucket_name)
    
    def upload_file(self, local_path: str, s3_key: str) -> None:
        """
        Upload file to S3.
        
        Args:
            local_path: Local file path
            s3_key: S3 object key
        """
        try:
            self.s3_client.upload_file(local_path, self.bucket_name, s3_key)
            print(f"Uploaded {local_path} to s3://{self.bucket_name}/{s3_key}")
        except ClientError as e:
            print(f"Error uploading file: {e}")
            raise
    
    def download_file(self, s3_key: str, local_path: str) -> None:
        """
        Download file from S3.
        
        Args:
            s3_key: S3 object key
            local_path: Local destination path
        """
        try:
            self.s3_client.download_file(self.bucket_name, s3_key, local_path)
            print(f"Downloaded s3://{self.bucket_name}/{s3_key} to {local_path}")
        except ClientError as e:
            print(f"Error downloading file: {e}")
            raise
    
    def upload_dataframe(
        self,
        df: pd.DataFrame,
        s3_key: str,
        file_format: str = 'parquet'
    ) -> None:
        """
        Upload DataFrame to S3.
        
        Args:
            df: DataFrame to upload
            s3_key: S3 object key
            file_format: 'parquet', 'csv', or 'json'
        """
        buffer = BytesIO()
        
        if file_format == 'parquet':
            df.to_parquet(buffer, index=False)
        elif file_format == 'csv':
            df.to_csv(buffer, index=False)
        elif file_format == 'json':
            df.to_json(buffer, orient='records', lines=True)
        else:
            raise ValueError(f"Unsupported format: {file_format}")
        
        buffer.seek(0)
        self.s3_client.put_object(
            Bucket=self.bucket_name,
            Key=s3_key,
            Body=buffer.getvalue()
        )
        print(f"Uploaded DataFrame to s3://{self.bucket_name}/{s3_key}")
    
    def read_dataframe(
        self,
        s3_key: str,
        file_format: str = 'parquet'
    ) -> pd.DataFrame:
        """
        Read DataFrame from S3.
        
        Args:
            s3_key: S3 object key
            file_format: 'parquet', 'csv', or 'json'
            
        Returns:
            DataFrame
        """
        obj = self.s3_client.get_object(Bucket=self.bucket_name, Key=s3_key)
        
        if file_format == 'parquet':
            return pd.read_parquet(BytesIO(obj['Body'].read()))
        elif file_format == 'csv':
            return pd.read_csv(BytesIO(obj['Body'].read()))
        elif file_format == 'json':
            return pd.read_json(BytesIO(obj['Body'].read()), lines=True)
        else:
            raise ValueError(f"Unsupported format: {file_format}")
    
    def list_objects(self, prefix: str = '') -> List[str]:
        """
        List objects in S3 with given prefix.
        
        Args:
            prefix: S3 key prefix
            
        Returns:
            List of object keys
        """
        objects = []
        paginator = self.s3_client.get_paginator('list_objects_v2')
        
        for page in paginator.paginate(Bucket=self.bucket_name, Prefix=prefix):
            if 'Contents' in page:
                objects.extend([obj['Key'] for obj in page['Contents']])
        
        return objects
    
    def delete_objects(self, s3_keys: List[str]) -> None:
        """
        Delete multiple objects from S3.
        
        Args:
            s3_keys: List of S3 object keys to delete
        """
        if not s3_keys:
            return
        
        objects = [{'Key': key} for key in s3_keys]
        
        # Delete in batches of 1000 (S3 limit)
        for i in range(0, len(objects), 1000):
            batch = objects[i:i+1000]
            self.s3_client.delete_objects(
                Bucket=self.bucket_name,
                Delete={'Objects': batch}
            )
        
        print(f"Deleted {len(s3_keys)} objects from S3")


# Example usage
if __name__ == "__main__":
    s3 = S3Manager('my-data-lake-bucket')
    
    # Upload CSV file
    s3.upload_file('data/users.csv', 'raw/users/2024-01-15/users.csv')
    
    # Read DataFrame from S3
    df = s3.read_dataframe('processed/users/users_cleaned.parquet')
    
    # List all parquet files in processed zone
    parquet_files = [
        f for f in s3.list_objects('processed/')
        if f.endswith('.parquet')
    ]
    print(f"Found {len(parquet_files)} parquet files")

AWS Glue - ETL Service

AWS Glue is a serverless ETL service. I use it for:

• Data Catalog: Central metadata repository

• ETL Jobs: Spark-based transformations

• Crawlers: Automatic schema discovery

# glue_operations.py
import boto3
from typing import Dict, Any, List

class GlueManager:
    """Manage AWS Glue operations"""
    
    def __init__(self, region: str = 'us-east-1'):
        """Initialize Glue manager"""
        self.glue_client = boto3.client('glue', region_name=region)
    
    def create_database(self, database_name: str, description: str = '') -> None:
        """
        Create Glue database.
        
        Args:
            database_name: Database name
            description: Database description
        """
        try:
            self.glue_client.create_database(
                DatabaseInput={
                    'Name': database_name,
                    'Description': description
                }
            )
            print(f"Created database: {database_name}")
        except self.glue_client.exceptions.AlreadyExistsException:
            print(f"Database {database_name} already exists")
    
    def create_crawler(
        self,
        crawler_name: str,
        database_name: str,
        s3_path: str,
        role_arn: str
    ) -> None:
        """
        Create Glue crawler for automatic schema discovery.
        
        Args:
            crawler_name: Crawler name
            database_name: Target database
            s3_path: S3 path to crawl
            role_arn: IAM role ARN for crawler
        """
        try:
            self.glue_client.create_crawler(
                Name=crawler_name,
                Role=role_arn,
                DatabaseName=database_name,
                Targets={
                    'S3Targets': [
                        {'Path': s3_path}
                    ]
                },
                SchemaChangePolicy={
                    'UpdateBehavior': 'UPDATE_IN_DATABASE',
                    'DeleteBehavior': 'LOG'
                }
            )
            print(f"Created crawler: {crawler_name}")
        except self.glue_client.exceptions.AlreadyExistsException:
            print(f"Crawler {crawler_name} already exists")
    
    def run_crawler(self, crawler_name: str) -> None:
        """
        Run Glue crawler.
        
        Args:
            crawler_name: Crawler name
        """
        self.glue_client.start_crawler(Name=crawler_name)
        print(f"Started crawler: {crawler_name}")
    
    def get_table_metadata(self, database_name: str, table_name: str) -> Dict[str, Any]:
        """
        Get table metadata from Glue catalog.
        
        Args:
            database_name: Database name
            table_name: Table name
            
        Returns:
            Table metadata dictionary
        """
        response = self.glue_client.get_table(
            DatabaseName=database_name,
            Name=table_name
        )
        return response['Table']
    
    def create_etl_job(
        self,
        job_name: str,
        role_arn: str,
        script_location: str,
        glue_version: str = '4.0',
        worker_type: str = 'G.1X',
        number_of_workers: int = 2
    ) -> None:
        """
        Create Glue ETL job.
        
        Args:
            job_name: Job name
            role_arn: IAM role ARN
            script_location: S3 path to PySpark script
            glue_version: Glue version
            worker_type: Worker node type
            number_of_workers: Number of workers
        """
        self.glue_client.create_job(
            Name=job_name,
            Role=role_arn,
            Command={
                'Name': 'glueetl',
                'ScriptLocation': script_location,
                'PythonVersion': '3'
            },
            GlueVersion=glue_version,
            WorkerType=worker_type,
            NumberOfWorkers=number_of_workers
        )
        print(f"Created ETL job: {job_name}")
    
    def start_job_run(self, job_name: str, arguments: Dict[str, str] = None) -> str:
        """
        Start Glue job run.
        
        Args:
            job_name: Job name
            arguments: Job arguments
            
        Returns:
            Job run ID
        """
        response = self.glue_client.start_job_run(
            JobName=job_name,
            Arguments=arguments or {}
        )
        job_run_id = response['JobRunId']
        print(f"Started job run: {job_run_id}")
        return job_run_id


# Example Glue PySpark script
GLUE_SCRIPT = """
# glue_etl_script.py
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

# Get job parameters
args = getResolvedOptions(sys.argv, ['JOB_NAME', 'SOURCE_PATH', 'TARGET_PATH'])

# Initialize Glue context
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read data from S3
datasource = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": [args['SOURCE_PATH']]},
    format="parquet"
)

# Transform data
transformed = datasource.apply_mapping([
    ("user_id", "long", "user_id", "long"),
    ("name", "string", "user_name", "string"),
    ("email", "string", "email", "string"),
    ("created_at", "timestamp", "registration_date", "timestamp")
])

# Filter out invalid records
filtered = Filter.apply(
    frame=transformed,
    f=lambda x: x["email"] is not None and "@" in x["email"]
)

# Write to S3 in partitioned format
glueContext.write_dynamic_frame.from_options(
    frame=filtered,
    connection_type="s3",
    connection_options={
        "path": args['TARGET_PATH'],
        "partitionKeys": ["registration_date"]
    },
    format="parquet"
)

job.commit()
"""

Amazon Redshift - Data Warehouse

Redshift is AWS's columnar data warehouse. I use it for:

• Analytics workloads: Complex queries on large datasets

• Business intelligence: Powering dashboards and reports

# redshift_operations.py
import psycopg2
from psycopg2 import sql
import pandas as pd
from typing import Dict, Any, Optional
from contextlib import contextmanager

class RedshiftManager:
    """Manage Amazon Redshift operations"""
    
    def __init__(
        self,
        host: str,
        port: int,
        database: str,
        user: str,
        password: str
    ):
        """
        Initialize Redshift manager.
        
        Args:
            host: Redshift cluster endpoint
            port: Port number (usually 5439)
            database: Database name
            user: Username
            password: Password
        """
        self.connection_params = {
            'host': host,
            'port': port,
            'database': database,
            'user': user,
            'password': password
        }
    
    @contextmanager
    def get_connection(self):
        """Get database connection with automatic cleanup"""
        conn = psycopg2.connect(**self.connection_params)
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()
    
    def execute_query(self, query: str) -> pd.DataFrame:
        """
        Execute SELECT query and return DataFrame.
        
        Args:
            query: SQL query
            
        Returns:
            Query results as DataFrame
        """
        with self.get_connection() as conn:
            return pd.read_sql(query, conn)
    
    def copy_from_s3(
        self,
        table_name: str,
        s3_path: str,
        iam_role: str,
        file_format: str = 'PARQUET',
        region: str = 'us-east-1'
    ) -> None:
        """
        Load data from S3 using COPY command (fastest method).
        
        Args:
            table_name: Target table
            s3_path: S3 path
            iam_role: IAM role ARN
            file_format: File format (PARQUET, CSV, JSON)
            region: AWS region
        """
        copy_query = f"""
            COPY {table_name}
            FROM '{s3_path}'
            IAM_ROLE '{iam_role}'
            FORMAT AS {file_format}
            REGION '{region}'
        """
        
        with self.get_connection() as conn:
            with conn.cursor() as cur:
                cur.execute(copy_query)
        
        print(f"Loaded data from {s3_path} to {table_name}")
    
    def unload_to_s3(
        self,
        query: str,
        s3_path: str,
        iam_role: str,
        file_format: str = 'PARQUET'
    ) -> None:
        """
        Unload query results to S3.
        
        Args:
            query: SELECT query
            s3_path: S3 destination path
            iam_role: IAM role ARN
            file_format: File format (PARQUET, CSV)
        """
        unload_query = f"""
            UNLOAD ('{query}')
            TO '{s3_path}'
            IAM_ROLE '{iam_role}'
            FORMAT AS {file_format}
            ALLOWOVERWRITE
            PARALLEL ON
        """
        
        with self.get_connection() as conn:
            with conn.cursor() as cur:
                cur.execute(unload_query)
        
        print(f"Unloaded data to {s3_path}")
    
    def vacuum_analyze(self, table_name: str) -> None:
        """
        Run VACUUM and ANALYZE on table for optimal performance.
        
        Args:
            table_name: Table to optimize
        """
        with self.get_connection() as conn:
            # VACUUM reclaims space and sorts data
            with conn.cursor() as cur:
                cur.execute(f"VACUUM {table_name}")
            
            # ANALYZE updates statistics
            with conn.cursor() as cur:
                cur.execute(f"ANALYZE {table_name}")
        
        print(f"Vacuumed and analyzed {table_name}")


# Example usage
if __name__ == "__main__":
    redshift = RedshiftManager(
        host='my-cluster.us-east-1.redshift.amazonaws.com',
        port=5439,
        database='analytics',
        user='admin',
        password='password'
    )
    
    # Load from S3 (fastest method for large datasets)
    redshift.copy_from_s3(
        table_name='fact_sales',
        s3_path='s3://my-bucket/data/sales/',
        iam_role='arn:aws:iam::123456789012:role/RedshiftRole',
        file_format='PARQUET'
    )
    
    # Query data
    df = redshift.execute_query("""
        SELECT 
            DATE_TRUNC('month', order_date) as month,
            SUM(total_amount) as revenue
        FROM fact_sales
        WHERE order_date >= '2024-01-01'
        GROUP BY 1
        ORDER BY 1
    """)
    
    print(df)

Snowflake

Snowflake is my preferred cloud data warehouse. It separates storage and compute, making it cost-effective and flexible.

Snowflake with Python

# snowflake_operations.py
import snowflake.connector
from snowflake.connector.pandas_tools import write_pandas
import pandas as pd
from typing import Dict, Any, Optional
from contextlib import contextmanager

class SnowflakeManager:
    """Manage Snowflake operations"""
    
    def __init__(
        self,
        account: str,
        user: str,
        password: str,
        warehouse: str,
        database: str,
        schema: str,
        role: Optional[str] = None
    ):
        """
        Initialize Snowflake manager.
        
        Args:
            account: Snowflake account identifier
            user: Username
            password: Password
            warehouse: Virtual warehouse name
            database: Database name
            schema: Schema name
            role: Role name (optional)
        """
        self.connection_params = {
            'account': account,
            'user': user,
            'password': password,
            'warehouse': warehouse,
            'database': database,
            'schema': schema
        }
        if role:
            self.connection_params['role'] = role
    
    @contextmanager
    def get_connection(self):
        """Get database connection with automatic cleanup"""
        conn = snowflake.connector.connect(**self.connection_params)
        try:
            yield conn
        finally:
            conn.close()
    
    def execute_query(self, query: str) -> pd.DataFrame:
        """
        Execute SELECT query and return DataFrame.
        
        Args:
            query: SQL query
            
        Returns:
            Query results as DataFrame
        """
        with self.get_connection() as conn:
            return pd.read_sql(query, conn)
    
    def execute_write(self, query: str) -> int:
        """
        Execute INSERT/UPDATE/DELETE query.
        
        Args:
            query: SQL query
            
        Returns:
            Number of rows affected
        """
        with self.get_connection() as conn:
            cursor = conn.cursor()
            result = cursor.execute(query)
            return result.rowcount
    
    def write_dataframe(
        self,
        df: pd.DataFrame,
        table_name: str,
        overwrite: bool = False
    ) -> None:
        """
        Write DataFrame to Snowflake table.
        
        Args:
            df: DataFrame to write
            table_name: Target table name
            overwrite: Whether to overwrite existing table
        """
        with self.get_connection() as conn:
            success, nchunks, nrows, _ = write_pandas(
                conn=conn,
                df=df,
                table_name=table_name,
                database=self.connection_params['database'],
                schema=self.connection_params['schema'],
                overwrite=overwrite
            )
            
            if success:
                print(f"Wrote {nrows} rows to {table_name}")
            else:
                raise Exception("Failed to write DataFrame to Snowflake")
    
    def copy_into_from_s3(
        self,
        table_name: str,
        s3_path: str,
        file_format: str,
        aws_access_key: str,
        aws_secret_key: str
    ) -> None:
        """
        Load data from S3 using COPY INTO.
        
        Args:
            table_name: Target table
            s3_path: S3 path (e.g., s3://bucket/path/)
            file_format: Snowflake file format name
            aws_access_key: AWS access key
            aws_secret_key: AWS secret key
        """
        copy_query = f"""
            COPY INTO {table_name}
            FROM '{s3_path}'
            CREDENTIALS = (
                AWS_KEY_ID = '{aws_access_key}'
                AWS_SECRET_KEY = '{aws_secret_key}'
            )
            FILE_FORMAT = (FORMAT_NAME = {file_format})
            ON_ERROR = 'CONTINUE'
        """
        
        rows_loaded = self.execute_write(copy_query)
        print(f"Loaded {rows_loaded} rows from S3 to {table_name}")
    
    def create_stage(
        self,
        stage_name: str,
        s3_path: str,
        aws_access_key: str,
        aws_secret_key: str
    ) -> None:
        """
        Create external stage for S3.
        
        Args:
            stage_name: Stage name
            s3_path: S3 bucket path
            aws_access_key: AWS access key
            aws_secret_key: AWS secret key
        """
        create_stage_query = f"""
            CREATE OR REPLACE STAGE {stage_name}
            URL = '{s3_path}'
            CREDENTIALS = (
                AWS_KEY_ID = '{aws_access_key}'
                AWS_SECRET_KEY = '{aws_secret_key}'
            )
        """
        
        self.execute_write(create_stage_query)
        print(f"Created stage: {stage_name}")
    
    def merge_upsert(
        self,
        target_table: str,
        source_table: str,
        merge_keys: list[str],
        update_columns: list[str]
    ) -> None:
        """
        Perform MERGE (upsert) operation.
        
        Args:
            target_table: Target table name
            source_table: Source table/CTE name
            merge_keys: Columns to match on
            update_columns: Columns to update
        """
        # Build merge conditions
        merge_condition = ' AND '.join([
            f"target.{key} = source.{key}" for key in merge_keys
        ])
        
        # Build update clause
        update_clause = ', '.join([
            f"{col} = source.{col}" for col in update_columns
        ])
        
        # Build insert columns
        all_columns = merge_keys + update_columns
        insert_columns = ', '.join(all_columns)
        insert_values = ', '.join([f"source.{col}" for col in all_columns])
        
        merge_query = f"""
            MERGE INTO {target_table} AS target
            USING {source_table} AS source
            ON {merge_condition}
            WHEN MATCHED THEN
                UPDATE SET {update_clause}
            WHEN NOT MATCHED THEN
                INSERT ({insert_columns})
                VALUES ({insert_values})
        """
        
        rows_affected = self.execute_write(merge_query)
        print(f"Merged {rows_affected} rows into {target_table}")


# Example usage
if __name__ == "__main__":
    sf = SnowflakeManager(
        account='abc12345.us-east-1',
        user='data_engineer',
        password='password',
        warehouse='COMPUTE_WH',
        database='ANALYTICS',
        schema='PUBLIC',
        role='DATA_ENGINEER'
    )
    
    # Query data
    df = sf.execute_query("""
        SELECT 
            DATE_TRUNC('month', order_date) as month,
            COUNT(*) as order_count,
            SUM(total_amount) as revenue
        FROM orders
        WHERE order_date >= '2024-01-01'
        GROUP BY 1
        ORDER BY 1
    """)
    
    print(df)
    
    # Write DataFrame
    new_data = pd.DataFrame({
        'user_id': [1, 2, 3],
        'name': ['Alice', 'Bob', 'Charlie']
    })
    
    sf.write_dataframe(new_data, 'users_staging', overwrite=True)

Databricks

Databricks provides a unified analytics platform built on Apache Spark. I use it for:

• Large-scale ETL: Processing terabytes of data

• Machine learning: MLflow integration

• Delta Lake: ACID transactions on data lakes

# databricks_operations.py
from databricks import sql
from databricks.sdk import WorkspaceClient
import pandas as pd
from typing import Optional

class DatabricksManager:
    """Manage Databricks operations"""
    
    def __init__(
        self,
        server_hostname: str,
        http_path: str,
        access_token: str
    ):
        """
        Initialize Databricks manager.
        
        Args:
            server_hostname: Databricks workspace hostname
            http_path: SQL warehouse HTTP path
            access_token: Personal access token
        """
        self.server_hostname = server_hostname
        self.http_path = http_path
        self.access_token = access_token
    
    def execute_query(self, query: str) -> pd.DataFrame:
        """
        Execute SQL query and return DataFrame.
        
        Args:
            query: SQL query
            
        Returns:
            Query results as DataFrame
        """
        with sql.connect(
            server_hostname=self.server_hostname,
            http_path=self.http_path,
            access_token=self.access_token
        ) as connection:
            with connection.cursor() as cursor:
                cursor.execute(query)
                
                # Fetch all results
                columns = [desc[0] for desc in cursor.description]
                data = cursor.fetchall()
                
                return pd.DataFrame(data, columns=columns)
    
    def write_to_delta(
        self,
        df: pd.DataFrame,
        table_name: str,
        mode: str = 'append'
    ) -> None:
        """
        Write DataFrame to Delta table using Spark.
        Note: This requires running in Databricks notebook/job.
        
        Args:
            df: DataFrame to write
            table_name: Target table name
            mode: Write mode ('append', 'overwrite', 'error', 'ignore')
        """
        # This would run in a Databricks notebook
        spark_code = f"""
        # Convert pandas DataFrame to Spark DataFrame
        spark_df = spark.createDataFrame(df)
        
        # Write to Delta table
        spark_df.write.format("delta").mode("{mode}").saveAsTable("{table_name}")
        """
        print(f"Write operation code:\n{spark_code}")


# Example PySpark notebook code for Databricks
DATABRICKS_NOTEBOOK = """
# Databricks notebook source
# MAGIC %md
# MAGIC # ETL Pipeline with Delta Lake

# COMMAND ----------
# Read from S3
df = spark.read.format("parquet").load("s3://my-bucket/raw/events/")

# COMMAND ----------
# Transform data
from pyspark.sql import functions as F

transformed_df = df \
    .filter(F.col("event_date") >= "2024-01-01") \
    .withColumn("event_hour", F.hour(F.col("event_timestamp"))) \
    .withColumn("user_type", 
        F.when(F.col("is_premium") == True, "premium")
         .otherwise("free")
    )

# COMMAND ----------
# Write to Delta Lake with partitioning
transformed_df.write \
    .format("delta") \
    .mode("append") \
    .partitionBy("event_date") \
    .saveAsTable("events_processed")

# COMMAND ----------
# MAGIC %md
# MAGIC # Query Delta table

# COMMAND ----------
# Query using SQL
spark.sql(\"\"\"
    SELECT 
        event_date,
        user_type,
        COUNT(*) as event_count
    FROM events_processed
    WHERE event_date >= current_date() - INTERVAL 7 DAYS
    GROUP BY event_date, user_type
    ORDER BY event_date DESC
\"\"\").display()

# COMMAND ----------
# Optimize Delta table (compaction and Z-ordering)
spark.sql(\"\"\"
    OPTIMIZE events_processed
    ZORDER BY (user_id, event_type)
\"\"\")

# COMMAND ----------
# Time travel - query historical version
spark.read \
    .format("delta") \
    .option("versionAsOf", 5) \
    .table("events_processed") \
    .display()
"""

Azure Data Services

Azure Data Factory

ADF is Azure's cloud ETL service, similar to AWS Glue:

# azure_data_factory.py
from azure.identity import DefaultAzureCredential
from azure.mgmt.datafactory import DataFactoryManagementClient
from azure.mgmt.datafactory.models import *

class AzureDataFactoryManager:
    """Manage Azure Data Factory operations"""
    
    def __init__(
        self,
        subscription_id: str,
        resource_group: str,
        factory_name: str
    ):
        """
        Initialize ADF manager.
        
        Args:
            subscription_id: Azure subscription ID
            resource_group: Resource group name
            factory_name: Data factory name
        """
        self.subscription_id = subscription_id
        self.resource_group = resource_group
        self.factory_name = factory_name
        
        # Authenticate
        credential = DefaultAzureCredential()
        self.client = DataFactoryManagementClient(credential, subscription_id)
    
    def create_pipeline(
        self,
        pipeline_name: str,
        activities: list
    ) -> None:
        """
        Create ADF pipeline.
        
        Args:
            pipeline_name: Pipeline name
            activities: List of pipeline activities
        """
        pipeline = PipelineResource(
            activities=activities
        )
        
        self.client.pipelines.create_or_update(
            self.resource_group,
            self.factory_name,
            pipeline_name,
            pipeline
        )
        
        print(f"Created pipeline: {pipeline_name}")
    
    def trigger_pipeline(self, pipeline_name: str, parameters: dict = None) -> str:
        """
        Trigger pipeline run.
        
        Args:
            pipeline_name: Pipeline name
            parameters: Pipeline parameters
            
        Returns:
            Run ID
        """
        run_response = self.client.pipelines.create_run(
            self.resource_group,
            self.factory_name,
            pipeline_name,
            parameters=parameters or {}
        )
        
        print(f"Triggered pipeline run: {run_response.run_id}")
        return run_response.run_id

Best Practices for Cloud Data Platforms

From my production experience:

1. Use managed services: Let cloud providers handle infrastructure—focus on data logic

2. Partition data: Partition by date or key dimensions for query performance

3. Choose the right format: Parquet for analytics, JSON for semi-structured data

4. Implement data lifecycle policies: Move old data to cheaper storage tiers

5. Monitor costs: Set up billing alerts and track resource usage

6. Use IAM roles: Avoid hardcoding credentials

7. Enable encryption: Encrypt data at rest and in transit

8. Test at scale: Cloud services behave differently with production data volumes

Key Takeaways

• S3/Azure Blob: Foundation for cloud data lakes

• Snowflake: Excellent for analytics with separation of storage/compute

• Databricks: Best for large-scale Spark processing and ML

• Managed ETL: Glue, Data Factory reduce operational overhead

• Cost optimization: Use partitioning, lifecycle policies, and right-sizing

• Security: Always use IAM roles and encrypt sensitive data

← Previous: SQL for Data Engineering | Next: Streaming & Real-Time Data →