AWS Graviton4 vs. AMD EPYC 9004: EC2 Instance Cost and Performance Comparison for 2026 Data Processing Workloads

import sys
import time
import logging
from argparse import ArgumentParser
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum, avg

# Configure logging for benchmark traceability
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

def create_spark_session(app_name: str, instance_type: str) -> SparkSession:
    """Initialize Spark session with Graviton4/EPYC optimized configs."""
    try:
        # Graviton4 benefits from ARM-optimized Spark native libraries
        # EPYC 9004 benefits from AVX-512 vectorized Parquet decoding
        builder = SparkSession.builder \
            .appName(app_name) \
            .config("spark.sql.parquet.enableVectorizedReader", "true") \
            .config("spark.sql.inMemoryColumnarStorage.compressed", "true")

        if "graviton" in instance_type.lower():
            builder.config("spark.sql.parquet.filterPushdown", "true") \
                   .config("spark.executor.extraJavaOptions", "-XX:+UseG1GC -XX:MaxGCPauseMillis=200")
        elif "epyc" in instance_type.lower():
            builder.config("spark.sql.parquet.vectorizedReader.batchSize", "4096") \
                   .config("spark.executor.extraJavaOptions", "-XX:+UseAVX512 -XX:+UseG1GC")

        return builder.getOrCreate()
    except Exception as e:
        logger.error(f"Failed to create Spark session: {e}")
        sys.exit(1)

def run_parquet_benchmark(spark: SparkSession, data_path: str, num_iterations: int = 3) -> float:
    """Run TPC-DS SF100 store_sales Parquet read benchmark, return avg rows/sec."""
    try:
        # Load 10TB TPC-DS SF100 store_sales Parquet dataset
        df = spark.read.parquet(data_path)
        logger.info(f"Loaded Parquet dataset with {df.count()} rows, schema: {df.printSchema()}")

        total_rows = 0
        total_time = 0.0

        for i in range(num_iterations):
            logger.info(f"Starting iteration {i+1}/{num_iterations}")
            start = time.time()

            # TPC-DS Q1: Calculate total sales per store, filter for 2026 data
            result = df.filter(col("ss_sold_date_sk") >= 18262) \  # 2026-01-01 in TPC-DS date key
                       .groupBy("ss_store_sk") \
                       .agg(
                           spark_sum("ss_sales_price").alias("total_sales"),
                           avg("ss_quantity").alias("avg_quantity")
                       ) \
                       .collect()

            elapsed = time.time() - start
            rows_processed = df.filter(col("ss_sold_date_sk") >= 18262).count()
            total_rows += rows_processed
            total_time += elapsed

            logger.info(f"Iteration {i+1}: Processed {rows_processed} rows in {elapsed:.2f}s ({rows_processed/elapsed:.0f} rows/sec)")

        avg_throughput = total_rows / total_time
        logger.info(f"Average throughput over {num_iterations} iterations: {avg_throughput:.0f} rows/sec")
        return avg_throughput
    except Exception as e:
        logger.error(f"Benchmark failed: {e}")
        sys.exit(1)

if __name__ == "__main__":
    parser = ArgumentParser(description="Spark Parquet throughput benchmark for Graviton4 vs EPYC 9004")
    parser.add_argument("--instance-type", required=True, help="EC2 instance type (e.g., r8g.4xlarge, m7a.4xlarge)")
    parser.add_argument("--data-path", required=True, help="S3 path to TPC-DS SF100 Parquet dataset")
    parser.add_argument("--iterations", type=int, default=3, help="Number of benchmark iterations")
    args = parser.parse_args()

    logger.info(f"Starting benchmark on instance type: {args.instance_type}")
    spark = create_spark_session("Graviton4_EPYC_Benchmark", args.instance_type)

    try:
        throughput = run_parquet_benchmark(spark, args.data_path, args.iterations)
        print(f"BENCHMARK_RESULT: {throughput:.0f} rows/sec")
    finally:
        spark.stop()

import boto3
import json
from datetime import datetime, timedelta
from argparse import ArgumentParser
from typing import Dict, List

# Initialize AWS clients for pricing and EC2
pricing_client = boto3.client("pricing", region_name="us-east-1")
ec2_client = boto3.client("ec2", region_name="us-east-1")

def get_on_demand_price(instance_type: str, region: str = "us-east-1") -> float:
    """Fetch on-demand hourly price for EC2 instance type via AWS Pricing API."""
    try:
        response = pricing_client.get_products(
            ServiceCode="AmazonEC2",
            Filters=[
                {"Type": "TERM_MATCH", "Field": "instanceType", "Value": instance_type},
                {"Type": "TERM_MATCH", "Field": "location", "Value": "US East (N. Virginia)"},
                {"Type": "TERM_MATCH", "Field": "operatingSystem", "Value": "Linux"},
                {"Type": "TERM_MATCH", "Field": "tenancy", "Value": "Shared"},
                {"Type": "TERM_MATCH", "Field": "preInstalledSw", "Value": "NA"}
            ],
            MaxResults=1
        )

        if not response["PriceList"]:
            raise ValueError(f"No pricing data found for {instance_type} in {region}")

        price_item = json.loads(response["PriceList"][0])
        terms = price_item["terms"]["OnDemand"]
        term_key = next(iter(terms))
        price_dimensions = terms[term_key]["priceDimensions"]
        price_key = next(iter(price_dimensions))
        hourly_price = float(price_dimensions[price_key]["pricePerUnit"]["USD"])

        return hourly_price
    except Exception as e:
        print(f"Error fetching price for {instance_type}: {e}")
        raise

def get_spot_discount(instance_type: str, region: str = "us-east-1") -> float:
    """Calculate average spot instance discount over last 30 days."""
    try:
        end_time = datetime.now()
        start_time = end_time - timedelta(days=30)

        response = ec2_client.describe_spot_price_history(
            InstanceTypes=[instance_type],
            StartTime=start_time,
            EndTime=end_time,
            ProductDescriptions=["Linux/UNIX"],
            AvailabilityZone=f"{region}a"
        )

        if not response["SpotPriceHistory"]:
            return 0.0

        on_demand = get_on_demand_price(instance_type, region)
        total_discount = 0.0
        count = 0

        for entry in response["SpotPriceHistory"]:
            spot_price = float(entry["SpotPrice"])
            discount = (on_demand - spot_price) / on_demand
            total_discount += discount
            count += 1

        return total_discount / count
    except Exception as e:
        print(f"Error calculating spot discount for {instance_type}: {e}")
        return 0.0

def calculate_tco(instance_type: str, num_instances: int, hours_per_month: int = 730) -> Dict:
    """Calculate 3-year TCO for cluster of instance types."""
    try:
        on_demand = get_on_demand_price(instance_type)
        spot_discount = get_spot_discount(instance_type)
        spot_hourly = on_demand * (1 - spot_discount)

        # 3-year cost: 12 months * 3 years, 730 hours/month
        monthly_on_demand = on_demand * hours_per_month * num_instances
        monthly_spot = spot_hourly * hours_per_month * num_instances

        annual_on_demand = monthly_on_demand * 12
        annual_spot = monthly_spot * 12

        tco_3yr_on_demand = annual_on_demand * 3
        tco_3yr_spot = annual_spot * 3

        return {
            "instance_type": instance_type,
            "on_demand_hourly": on_demand,
            "spot_hourly": spot_hourly,
            "spot_discount_pct": spot_discount * 100,
            "monthly_on_demand": monthly_on_demand,
            "monthly_spot": monthly_spot,
            "tco_3yr_on_demand": tco_3yr_on_demand,
            "tco_3yr_spot": tco_3yr_spot
        }
    except Exception as e:
        print(f"TCO calculation failed: {e}")
        raise

if __name__ == "__main__":
    parser = ArgumentParser(description="EC2 TCO Calculator for Graviton4 vs EPYC 9004")
    parser.add_argument("--graviton-type", default="r8g.4xlarge", help="Graviton4 instance type")
    parser.add_argument("--epyc-type", default="m7a.4xlarge", help="AMD EPYC 9004 instance type")
    parser.add_argument("--num-instances", type=int, default=100, help="Number of instances in cluster")
    args = parser.parse_args()

    print("Calculating 3-year TCO for data processing cluster...")
    graviton_tco = calculate_tco(args.graviton_type, args.num_instances)
    epyc_tco = calculate_tco(args.epyc_type, args.num_instances)

    print("\n=== Graviton4 TCO ===")
    for k, v in graviton_tco.items():
        print(f"{k}: {v:.2f}" if isinstance(v, float) else f"{k}: {v}")

    print("\n=== AMD EPYC 9004 TCO ===")
    for k, v in epyc_tco.items():
        print(f"{k}: {v:.2f}" if isinstance(v, float) else f"{k}: {v}")

    savings = graviton_tco["tco_3yr_spot"] - epyc_tco["tco_3yr_spot"]
    print(f"\n3-Year Spot TCO Savings with Graviton4: ${abs(savings):.2f}")

import psycopg2
import time
import logging
from argparse import ArgumentParser
from typing import List, Tuple

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

def init_pgbench_db(conn_str: str, scale_factor: int = 100) -> None:
    """Initialize pgbench database with specified scale factor (100 = ~1.5GB data)."""
    try:
        logger.info(f"Initializing pgbench database with scale factor {scale_factor}")
        import subprocess
        result = subprocess.run(
            ["pgbench", "-i", "-s", str(scale_factor), conn_str],
            check=True,
            capture_output=True,
            text=True
        )
        logger.info(f"pgbench init complete: {result.stdout}")
    except subprocess.CalledProcessError as e:
        logger.error(f"pgbench init failed: {e.stderr}")
        raise
    except Exception as e:
        logger.error(f"DB init failed: {e}")
        raise

def run_oltp_benchmark(conn_str: str, instance_type: str, duration_sec: int = 300) -> float:
    """Run pgbench OLTP benchmark, return average TPS."""
    try:
        # Graviton4 uses ARM-optimized PostgreSQL builds from AWS
        # EPYC 9004 uses AVX-512 optimized PostgreSQL 16 builds
        cmd = [
            "pgbench",
            "-c", "16",  # Match vCPU count of 4xlarge instances
            "-j", "4",
            "-T", str(duration_sec),
            "-P", "10",  # Report every 10 seconds
            conn_str
        ]

        logger.info(f"Running pgbench on {instance_type} for {duration_sec}s")
        result = subprocess.run(
            cmd,
            check=True,
            capture_output=True,
            text=True
        )

        # Parse TPS from pgbench output
        tps_lines = [line for line in result.stdout.split("\n") if "tps" in line.lower()]
        if not tps_lines:
            raise ValueError("No TPS data found in pgbench output")

        # Extract average TPS from final line
        final_line = tps_lines[-1]
        tps = float(final_line.split("tps")[0].split("=")[-1].strip())
        logger.info(f"Average TPS for {instance_type}: {tps:.0f}")
        return tps
    except Exception as e:
        logger.error(f"Benchmark failed: {e}")
        raise

def compare_instances(graviton_conn: str, epyc_conn: str, scale: int = 100, duration: int = 300) -> Tuple[float, float]:
    """Run benchmarks on both instances, return (graviton_tps, epyc_tps)."""
    try:
        init_pgbench_db(graviton_conn, scale)
        graviton_tps = run_oltp_benchmark(graviton_conn, "Graviton4 r8g.4xlarge", duration)

        init_pgbench_db(epyc_conn, scale)
        epyc_tps = run_oltp_benchmark(epyc_conn, "AMD EPYC 9004 m7a.4xlarge", duration)

        return graviton_tps, epyc_tps
    except Exception as e:
        logger.error(f"Comparison failed: {e}")
        raise

if __name__ == "__main__":
    parser = ArgumentParser(description="PostgreSQL 16 OLTP Benchmark: Graviton4 vs EPYC 9004")
    parser.add_argument("--graviton-conn", required=True, help="Graviton4 PostgreSQL connection string")
    parser.add_argument("--epyc-conn", required=True, help="EPYC 9004 PostgreSQL connection string")
    parser.add_argument("--scale", type=int, default=100, help="pgbench scale factor")
    parser.add_argument("--duration", type=int, default=300, help="Benchmark duration in seconds")
    args = parser.parse_args()

    graviton_tps, epyc_tps = compare_instances(
        args.graviton_conn,
        args.epyc_conn,
        args.scale,
        args.duration
    )

    print(f"\n=== Benchmark Results ===")
    print(f"Graviton4 r8g.4xlarge TPS: {graviton_tps:.0f}")
    print(f"AMD EPYC 9004 m7a.4xlarge TPS: {epyc_tps:.0f}")
    print(f"EPYC outperforms Graviton4 by {(epyc_tps - graviton_tps)/graviton_tps*100:.1f}% for OLTP")

# User data script to validate ARM64 runtime and install Temurin JDK 21
#!/bin/bash
ARCH=$(uname -m)
if [ "$ARCH" != "aarch64" ]; then
  echo "ERROR: Expected aarch64, got $ARCH. Terminating instance."
  shutdown -h now
fi
wget https://github.com/adoptium/temurin21-binaries/releases/download/jdk-21.0.2%2B13/OpenJDK21U-jdk_aarch64_linux_hotspot_21.0.2_13.tar.gz
tar -xzf OpenJDK21U-jdk_aarch64_linux_hotspot_21.0.2_13.tar.gz -C /usr/lib/jvm/
export JAVA_HOME=/usr/lib/jvm/jdk-21.0.2+13
export PATH=$JAVA_HOME/bin:$PATH
java -version  # Validate ARM64 JDK

# CMake configuration for EPYC 9004 optimized C++ HPC workload
cmake_minimum_required(VERSION 3.28)
project(EPYC_HPC_Benchmark)
set(CMAKE_CXX_STANDARD 20)
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -march=znver4 -mavx512f -O3 -flto")
find_package(AOCL REQUIRED COMPONENTS BLAS LAPACK)
add_executable(benchmark main.cpp)
target_link_libraries(benchmark PRIVATE AOCL::BLAS AOCL::LAPACK)

# Spark configuration for spot instance checkpointing
spark = SparkSession.builder \
    .config("spark.sql.streaming.checkpointLocation", "s3a://my-bucket/checkpoints/") \
    .config("spark.streaming.backpressure.enabled", "true") \
    .config("spark.streaming.kafka.maxRatePerPartition", "1000") \
    .getOrCreate()