Data Engineering on AWS: Complete Guide

Discover the AWS services for processing and analyzing data at scale.

1. Data Ingestion with Amazon Kinesis

Stream Configuration

# Kinesis configuration example
import boto3

def setup_kinesis_stream():
    kinesis = boto3.client('kinesis')
    
    # Create the stream
    stream = kinesis.create_stream(
        StreamName='MonStream',
        ShardCount=2
    )
    
    # Wait for the stream to be created
    waiter = kinesis.get_waiter('stream_exists')
    waiter.wait(StreamName='MonStream')
    
    return stream['StreamName']

Sending Data

# Example of sending data to Kinesis
def send_data_to_kinesis():
    kinesis = boto3.client('kinesis')
    
    data = {
        'timestamp': '2024-03-22T10:00:00',
        'value': 42,
        'type': 'metric'
    }
    
    response = kinesis.put_record(
        StreamName='MonStream',
        Data=json.dumps(data),
        PartitionKey='metric'
    )
    
    return response['ShardId']

2. Processing with Amazon EMR

Cluster Configuration

# EMR configuration example
def setup_emr_cluster():
    emr = boto3.client('emr')
    
    cluster_config = {
        'Name': 'MonClusterEMR',
        'ReleaseLabel': 'emr-6.10.0',
        'Applications': [
            {'Name': 'Spark'},
            {'Name': 'Hadoop'}
        ],
        'Instances': {
            'InstanceGroups': [
                {
                    'Name': 'Master',
                    'Market': 'ON_DEMAND',
                    'InstanceRole': 'MASTER',
                    'InstanceType': 'm5.xlarge',
                    'InstanceCount': 1
                },
                {
                    'Name': 'Core',
                    'Market': 'ON_DEMAND',
                    'InstanceRole': 'CORE',
                    'InstanceType': 'm5.xlarge',
                    'InstanceCount': 2
                }
            ],
            'KeepJobFlowAliveWhenNoSteps': True,
            'TerminationProtected': False
        }
    }
    
    response = emr.run_job_flow(**cluster_config)
    return response['JobFlowId']

Running a Spark Job

# Example of running a Spark job
def run_spark_job():
    emr = boto3.client('emr')
    
    step_config = {
        'Name': 'MonJobSpark',
        'ActionOnFailure': 'CONTINUE',
        'HadoopJarStep': {
            'Jar': 'command-runner.jar',
            'Args': [
                'spark-submit',
                '--class', 'com.example.MainClass',
                's3://mon-bucket/jobs/mon-job.jar'
            ]
        }
    }
    
    response = emr.add_job_flow_steps(
        JobFlowId='j-XXXXX',
        Steps=[step_config]
    )
    
    return response['StepIds']

3. Storage with Amazon S3 and Amazon Redshift

Redshift Configuration

# Redshift configuration example
def setup_redshift_cluster():
    redshift = boto3.client('redshift')
    
    cluster_config = {
        'ClusterIdentifier': 'mon-cluster',
        'NodeType': 'dc2.large',
        'NumberOfNodes': 2,
        'MasterUsername': 'admin',
        'MasterUserPassword': 'password123',
        'DBName': 'mon_database',
        'ClusterType': 'multi-node',
        'VpcSecurityGroupIds': ['sg-xxxxx'],
        'ClusterSubnetGroupName': 'mon-subnet-group'
    }
    
    response = redshift.create_cluster(**cluster_config)
    return response['Cluster']['ClusterIdentifier']

Loading Data

# Example of loading data into Redshift
def load_data_to_redshift():
    redshift = boto3.client('redshift-data')
    
    query = """
    COPY mon_table
    FROM 's3://mon-bucket/data/'
    IAM_ROLE 'arn:aws:iam::account:role/RedshiftCopyRole'
    CSV
    """
    
    response = redshift.execute_statement(
        ClusterIdentifier='mon-cluster',
        Database='mon_database',
        DbUser='admin',
        Sql=query
    )
    
    return response['Id']

4. Transformation with AWS Glue

Job Configuration

# Glue configuration example
def setup_glue_job():
    glue = boto3.client('glue')
    
    job_config = {
        'Name': 'MonJobGlue',
        'Role': 'arn:aws:iam::account:role/AWSGlueServiceRole',
        'Command': {
            'Name': 'glueetl',
            'ScriptLocation': 's3://mon-bucket/scripts/mon-script.py',
            'PythonVersion': '3'
        },
        'DefaultArguments': {
            '--job-language': 'python',
            '--TempDir': 's3://mon-bucket/temp/',
            '--job-bookmark-option': 'job-bookmark-enable'
        },
        'MaxRetries': 0,
        'Timeout': 2880,
        'MaxCapacity': 10.0
    }
    
    response = glue.create_job(**job_config)
    return response['Name']

ETL Script

# Glue ETL script example
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame

def main():
    # Initialize the context
    glueContext = GlueContext(SparkContext.getOrCreate())
    
    # Create the DynamicFrames
    source_dyf = glueContext.create_dynamic_frame.from_catalog(
        database="mon_database",
        table_name="source_table"
    )
    
    # Transform the data
    transformed_dyf = source_dyf.map(
        lambda x: {
            'id': x['id'],
            'value': x['value'] * 2,
            'timestamp': x['timestamp']
        }
    )
    
    # Write the results
    glueContext.write_dynamic_frame.from_options(
        frame=transformed_dyf,
        connection_type="s3",
        connection_options={
            "path": "s3://mon-bucket/output/"
        },
        format="parquet"
    )

if __name__ == "__main__":
    main()

5. Visualization with Amazon QuickSight

Dataset Configuration

# QuickSight configuration example
def setup_quicksight_dataset():
    quicksight = boto3.client('quicksight')
    
    dataset_config = {
        'AwsAccountId': '123456789012',
        'DataSetId': 'mon-dataset',
        'Name': 'MonDataset',
        'PhysicalTableMap': {
            'Table1': {
                'S3Source': {
                    'DataSourceArn': 'arn:aws:quicksight:region:account:datasource/mon-source',
                    'InputColumns': [
                        {
                            'Name': 'id',
                            'Type': 'STRING'
                        },
                        {
                            'Name': 'value',
                            'Type': 'INTEGER'
                        }
                    ]
                }
            }
        },
        'LogicalTableMap': {
            'Table1': {
                'Alias': 'MonTable',
                'DataTransforms': [],
                'Source': {
                    'PhysicalTableId': 'Table1'
                }
            }
        }
    }
    
    response = quicksight.create_data_set(**dataset_config)
    return response['DataSetId']

Creating a Dashboard

# Example of creating a QuickSight dashboard
def create_quicksight_dashboard():
    quicksight = boto3.client('quicksight')
    
    dashboard_config = {
        'AwsAccountId': '123456789012',
        'DashboardId': 'mon-dashboard',
        'Name': 'MonDashboard',
        'SourceEntity': {
            'SourceTemplate': {
                'DataSetReferences': [
                    {
                        'DataSetPlaceholder': 'MonDataset',
                        'DataSetArn': 'arn:aws:quicksight:region:account:dataset/mon-dataset'
                    }
                ],
                'Arn': 'arn:aws:quicksight:region:account:template/mon-template'
            }
        }
    }
    
    response = quicksight.create_dashboard(**dashboard_config)
    return response['DashboardId']

Conclusion

Key takeaways:

• Ingest real-time data with Kinesis
• Process data with EMR
• Store and analyze with Redshift
• Transform with Glue
• Visualize with QuickSight

Recommendations:

• Choose the right services for your needs
• Optimize processing costs
• Build robust ETL pipelines
• Monitor performance
• Secure your data