#!/bin/bash
# EBS optimization script for data processing instances
set -euo pipefail

# Update kernel I/O scheduler for SSDs
echo none | sudo tee /sys/block/nvme0n1/queue/scheduler

# Increase read-ahead for sequential workloads
sudo blockdev --setra 4096 /dev/nvme0n1

# Configure sysctl for optimal I/O performance
sudo sysctl -w vm.dirty_ratio=40
sudo sysctl -w vm.dirty_background_ratio=10
sudo sysctl -w vm.dirty_expire_centisecs=3000
sudo sysctl -w vm.dirty_writeback_centisecs=500

echo "EBS optimization applied successfully"

import boto3

ec2 = boto3.client('ec2')

# Get spot instance pricing history
response = ec2.describe_spot_price_history(
    InstanceTypes=['r5.4xlarge', 'r5.8xlarge', 'c5.4xlarge'],
    ProductDescriptions=['Linux/UNIX'],
    StartTime=datetime.now() - timedelta(days=7),
    EndTime=datetime.now()
)

for price in response['SpotPriceHistory']:
    print(f"{price['InstanceType']} in {price['AvailabilityZone']}: ${price['SpotPrice']}/hr")

{
  "SpotFleetRequestConfig": {
    "AllocationStrategy": "capacityOptimized",
    "IamFleetRole": "arn:aws:iam::role/spot-fleet-role",
    "TargetCapacitySpecification": {
      "DefaultTargetCapacityType": "spot",
      "TotalTargetCapacity": 100
    },
    "LaunchTemplateConfigs": [
      {
        "LaunchTemplateSpecification": {
          "LaunchTemplateId": "lt-0123456789abcdef0",
          "Version": "$Latest"
        },
        "Overrides": [
          {"InstanceType": "r5.4xlarge", "SubnetId": "subnet-aaa"},
          {"InstanceType": "r5.8xlarge", "SubnetId": "subnet-bbb"},
          {"InstanceType": "c5.4xlarge", "SubnetId": "subnet-ccc"}
        ]
      }
    ],
    "SpotMaintenanceStrategies": {
      "CapacityRebalance": {
        "ReplacementStrategy": "launch-before-terminate"
      }
    }
  }
}

import boto3

autoscaling = boto3.client('autoscaling')

# Create auto scaling group for EMR task nodes
autoscaling.create_auto_scaling_group(
    AutoScalingGroupName='emr-task-nodes',
    LaunchTemplate={
        'LaunchTemplateId': 'lt-0123456789abcdef0',
        'Version': '$Latest'
    },
    MinSize=0,
    MaxSize=20,
    DesiredCapacity=4,
    AvailabilityZones=['us-east-1a', 'us-east-1b', 'us-east-1c'],
    TargetGroupARNs=['arn:aws:elasticloadbalancing:us-east-1:123456789:targetgroup/emr-tg/abc'],
    HealthCheckType='EC2',
    HealthCheckGracePeriod=300,
    Tags=[
        {
            'Key': 'Name',
            'Value': 'EMR-Task-Node',
            'PropagateAtLaunch': True
        }
    ]
)

# Scaling policy based on YARN pending containers
autoscaling.put_scaling_policy(
    AutoScalingGroupName='emr-task-nodes',
    PolicyName='yarn-pending-scaling',
    PolicyType='TargetTrackingScaling',
    TargetTrackingConfiguration={
        'PredefinedMetricSpecification': {
            'PredefinedMetricType': 'ASGAverageCPUUtilization'
        },
        'TargetValue': 70.0,
        'ScaleInCooldown': 300,
        'ScaleOutCooldown': 60
    }
)

#!/bin/bash
# Build custom AMI for data engineering workloads
set -euo pipefail

# Update system packages
sudo yum update -y

# Install Java 11 (required for Spark/Hadoop)
sudo amazon-linux-extras install java11 -y

# Install Spark
wget https://archive.apache.org/dist/spark/spark-3.4.1/spark-3.4.1-bin-hadoop3.tgz
tar -xzf spark-3.4.1-bin-hadoop3.tgz -C /opt/
echo 'export SPARK_HOME=/opt/spark-3.4.1-bin-hadoop3' >> /etc/profile.d/spark.sh
echo 'export PATH=$SPARK_HOME/bin:$PATH' >> /etc/profile.d/spark.sh

# Install Python data libraries
pip3 install pandas pyspark boto3 pyarrow fastparquet

# Configure JVM for Spark
cat > /opt/spark/conf/spark-defaults.conf << 'EOF'
spark.executor.memory=8g
spark.driver.memory=4g
spark.sql.shuffle.partitions=200
spark.serializer=org.apache.spark.serializer.KryoSerializer
EOF

# Optimize OS settings
echo 'vm.swappiness=10' >> /etc/sysctl.conf
echo 'net.core.rmem_max=16777216' >> /etc/sysctl.conf
echo 'net.core.wmem_max=16777216' >> /etc/sysctl.conf

# Clean up
sudo yum clean all
sudo rm -rf /var/cache/yum

import boto3

ec2 = boto3.client('ec2')

# Attach additional ENI for high-bandwidth workloads
response = ec2.create_network_interface(
    SubnetId='subnet-0123456789abcdef0',
    Description='High-bandwidth data interface',
    Groups=['sg-0123456789abcdef0'],
    TagSpecifications=[
        {
            'ResourceType': 'network-interface',
            'Tags': [
                {'Key': 'Name', 'Value': 'DataPipeline-HighBandwidth'},
                {'Key': 'Purpose', 'Value': 'Spark Shuffle Traffic'}
            ]
        }
    ]
)
eni_id = response['NetworkInterface']['NetworkInterfaceId']

# Attach ENI to existing instance
ec2.attach_network_interface(
    NetworkInterfaceId=eni_id,
    InstanceId='i-0123456789abcdef0',
    DeviceIndex=1
)

# Enable VPC Flow Logs for data engineering VPC
logs_client = boto3.client('logs')
ec2_client = boto3.client('ec2')

# Create flow log
ec2_client.create_flow_logs(
    ResourceIds=['vpc-0123456789abcdef0'],
    ResourceType='VPC',
    TrafficType='ALL',
    LogDestinationType='cloud-watch-logs',
    LogGroupName='/vpc/flowlogs/data-engineering',
    DeliverLogsPermissionArn='arn:aws:iam::role/FlowLogsRole',
    MaxAggregationInterval=60
)

import boto3
import signal
import sys

ec2 = boto3.client('ec2')

def handle_termination(signum, frame):
    """Handle Spot termination notice gracefully."""
    # Get termination notice from instance metadata
    import urllib.request
    try:
        response = urllib.request.urlopen(
            'http://169.254.169.254/latest/meta-data/spot/instance-action',
            timeout=1
        )
        action = json.loads(response.read())
        print(f"Termination notice received: {action}")
        
        # Checkpoint current progress to S3
        checkpoint_to_s3()
        
        # Deregister from load balancer
        deregister_from_target_group()
        
        # Flush any buffered writes
        flush_buffers()
        
    except Exception as e:
        print(f"Error during termination handling: {e}")
    finally:
        sys.exit(0)

# Register signal handler
signal.signal(signal.SIGTERM, handle_termination)

def checkpoint_to_s3():
    """Save current processing state to S3."""
    s3 = boto3.client('s3')
    state = {
        'processed_offset': current_offset,
        'last_record_id': last_record_id,
        'timestamp': datetime.utcnow().isoformat()
    }
    s3.put_object(
        Bucket='checkpoint-bucket',
        Key=f'checkpoints/{instance_id}/state.json',
        Body=json.dumps(state)
    )

#!/bin/bash
# Monitor instance health and publish to CloudWatch
INSTANCE_ID=$(curl -s http://169.254.169.254/latest/meta-data/instance-id)
REGION=$(curl -s http://169.254.169.254/latest/meta-data/placement/region)

# Get memory usage
MEMORY_USED=$(free -m | awk 'NR==2{printf "%.2f", $3*100/$2}')
MEMORY_TOTAL=$(free -m | awk 'NR==2{print $2}')

# Get disk usage
DISK_USED=$(df -h / | awk 'NR==2{print $5}' | tr -d '%')
DISK_TOTAL=$(df -h / | awk 'NR==2{print $2}')

# Get load average
LOAD_1=$(cat /proc/loadavg | awk '{print $1}')
LOAD_5=$(cat /proc/loadavg | awk '{print $2}')
LOAD_15=$(cat /proc/loadavg | awk '{print $3}')

# Publish to CloudWatch
aws cloudwatch put-metric-data \
    --region $REGION \
    --namespace "EC2/DataEngineering" \
    --metric-data \
        "MetricName=MemoryUtilization,Dimensions=[{Name=InstanceId,Value=$INSTANCE_ID}],Value=$MEMORY_USED,Unit=Percent" \
        "MetricName=DiskUtilization,Dimensions=[{Name=InstanceId,Value=$INSTANCE_ID}],Value=$DISK_USED,Unit=Percent" \
        "MetricName=LoadAverage1Min,Dimensions=[{Name=InstanceId,Value=$INSTANCE_ID}],Value=$LOAD_1,Unit=None"

echo "Health metrics published at $(date)"