scripts/evaluation.py

# Load Presto and Spark data
import pandas as pd

from utility import human_readable_size

print("======= # of jobs and traffic volume ========")
presto_df = pd.read_csv("../metrics_per_day_presto.csv", parse_dates=['date'])
spark_df = pd.read_csv("../metrics_per_day_spark.csv", parse_dates=['date'])

# Compute totals
presto_total_jobs = presto_df['daily_jobs'].sum()
spark_total_jobs = spark_df['daily_jobs'].sum()

presto_total_rw_bytes = presto_df['daily_read_volume'].sum()
spark_total_rw_bytes = spark_df['daily_read_volume'].sum() + spark_df['daily_write_volume'].sum()

print(f"Presto: {presto_total_jobs:,} jobs, {presto_total_rw_bytes / 1024 ** 5:.2f} PB read")
print(f"Spark: {spark_total_jobs:,} jobs, {spark_total_rw_bytes / 1024 ** 5:.2f} PB read+write")
print(f"Total jobs: {presto_total_jobs + spark_total_jobs:,} jobs")
print(f"Total traffic volume: {(presto_total_rw_bytes + spark_total_rw_bytes) / 1024 ** 6:.2f} EB read+write")

print("======= Sec1 Overall Statistics ========")
# header: tag,c,network_cost, egress_cost,rep_cost,
# total_cost,total_cost_std,ingress_volume,egress_volume
df = pd.read_csv("../overall_stats.csv")

# Filter data for c = 50
df_c50 = df[df['c'] == 50]

# Get Moirai's data and all unique tags
moirai_data = df_c50[df_c50['tag'] == 'Moirai']
print("Moirai data", moirai_data[['c', 'network_cost', 'egress_cost', 'rep_cost']])


# Loop through each tag and calculate the improvement
for tag in ['Yugong', 'No\nRep', 'Rep\n3Mon.']:
    # Get the data for the current tag
    tag_data = df_c50[df_c50['tag'] == tag]

    # Merge tag data with Moirai data based on index to align rows
    merged_data = pd.merge(tag_data,
                           moirai_data,
                           on='c', suffixes=('_tag', '_moirai'))
    # Calculate the percentage improvement for each cost metric
    merged_data['total'] = (merged_data['total_cost_tag'] - merged_data[
        'total_cost_moirai']) / merged_data['total_cost_tag'] * 100
    merged_data['network'] = (merged_data['network_cost_tag'] - merged_data[
        'network_cost_moirai']) / merged_data['network_cost_tag'] * 100
    merged_data['egress'] = (merged_data['egress_cost_tag'] - merged_data[
        'egress_cost_moirai']) / merged_data['egress_cost_tag'] * 100
    merged_data['rep'] = (merged_data['rep_cost_tag'] - merged_data[
        'rep_cost_moirai']) / merged_data['rep_cost_tag'] * 100

    print(tag)
    print(merged_data[['total', 'network', 'egress', 'rep']])

print("======= Sec5 Overall Statistics ========")
# header: tag,c,network_cost, egress_cost,rep_cost,
# total_cost,total_cost_std,ingress_volume,egress_volume
df = pd.read_csv("../overall_stats.csv")
moirai_data = df[df['tag'] == 'Moirai']

print("** how Moirai improves over other approaches in Sec5 **")
# Loop through each tag and calculate the improvement
for tag in ['Volley', 'No\nRep', 'Rep\n3Mon.', 'Rep\nTop2.5%', 'Yugong']:
    # Get the data for the current tag
    tag_data = df[df['tag'] == tag]

    # Merge tag data with Moirai data based on index to align rows
    merged_data = pd.merge(tag_data,
                           moirai_data,
                           on='c', suffixes=('_tag', '_moirai'))
    # Calculate the percentage improvement for each cost metric
    merged_data['total'] = (merged_data['total_cost_tag'] - merged_data[
        'total_cost_moirai']) / merged_data['total_cost_tag'] * 100
    merged_data['traffic'] = (merged_data['ingress_volume_tag'] + merged_data['egress_volume_tag'] - \
                            (merged_data['ingress_volume_moirai'] + merged_data['egress_volume_moirai'])) / \
                            (merged_data['ingress_volume_tag'] + merged_data['egress_volume_tag']) * 100
    merged_data['network'] = (merged_data['network_cost_tag'] - merged_data[
        'network_cost_moirai']) / merged_data['network_cost_tag'] * 100
    merged_data['egress'] = (merged_data['egress_cost_tag'] - merged_data[
        'egress_cost_moirai']) / merged_data['egress_cost_tag'] * 100
    merged_data['rep'] = (merged_data['rep_cost_tag'] - merged_data[
        'rep_cost_moirai']) / merged_data['rep_cost_tag'] * 100

    print(tag)
    print(merged_data[['c', 'total', 'traffic', 'network', 'egress', 'rep']].round(2))

print("Reduction of Volley vs No Rep:")
merged_data = pd.merge(df[df['tag'] == 'No\nRep'], df[df['tag'] == 'Volley'],
                           on='c', suffixes=('_norep', '_volley'))
merged_data['total'] = (merged_data['total_cost_norep'] - merged_data['total_cost_volley']) / \
    merged_data['total_cost_volley'] * 100

print(merged_data[['c', 'total']].round(1))



print("** gap between other approaches and Moirai **")
# Loop through each tag and calculate the improvement
for tag in ['Volley', 'No\nRep', 'Rep\n3Mon.', 'Rep\nTop2.5%', 'Yugong', 'Moi\nJobDist', 'Moi\n1%Job']:
    # Get the data for the current tag
    tag_data = df[df['tag'] == tag]

    # Merge tag data with Moirai data based on index to align rows
    merged_data = pd.merge(tag_data,
                           moirai_data,
                           on='c', suffixes=('_tag', '_moirai'))
    # Calculate the percentage improvement for each cost metric
    merged_data['total'] = (merged_data['total_cost_tag'] - merged_data[
        'total_cost_moirai']) / merged_data['total_cost_moirai'] * 100
    merged_data['traffic'] = (merged_data['ingress_volume_tag'] + merged_data['egress_volume_tag'] - \
                            (merged_data['ingress_volume_moirai'] + merged_data['egress_volume_moirai'])) / \
                            (merged_data['ingress_volume_moirai'] + merged_data['egress_volume_moirai']) * 100
    merged_data['network'] = (merged_data['network_cost_tag'] - merged_data[
        'network_cost_moirai']) / merged_data['network_cost_moirai'] * 100
    merged_data['egress'] = (merged_data['egress_cost_tag'] - merged_data[
        'egress_cost_moirai']) / merged_data['egress_cost_moirai'] * 100
    merged_data['rep'] = (merged_data['rep_cost_tag'] - merged_data[
        'rep_cost_moirai']) / merged_data['rep_cost_moirai'] * 100

    print(tag)
    print(merged_data[['c', 'total', 'traffic', 'network', 'egress', 'rep']].round(0).astype(int))

print("========= Moirai Data Movement weekly =========")
# period,mode,cloud_computation_target,movement_ingress_bytes,movement_egress_bytes
df = pd.read_csv("../sample_1.000_rep0.002/log.csv")
df = df[df['mode'] == 'size-predict']
df['movement_bytes'] = df['movement_ingress_bytes'] + df['movement_egress_bytes']
df['movement_egress_cost'] = df['movement_egress_bytes'] * 0.02 / 1024 ** 3  # egress cost $0.02 per GB

# Group and process
grouped = df.groupby('cloud_computation_target')['movement_bytes'].apply(list)
# Print sorted values per target in human-readable form
for target, values in grouped.items():
    sorted_human_readable = [human_readable_size(v) for v in sorted(values)]
    print(f"{target}: {sorted_human_readable}")

# Group and process
grouped = df.groupby('cloud_computation_target')['movement_egress_cost'].apply(list)
# Print sorted values per target in human-readable form
for target, values in grouped.items():
    sorted_human_readable = [round(v, 0) for v in sorted(values)]
    print(f"{target}: {sorted_human_readable}")

print("========= Moirai Job Routing =========")
# header: period,mode,cloud_computation_target,
# ingress_byte_Presto,egress_byte_Presto,
# ingress_byte_Spark,egress_byte_Spark
df = pd.read_csv("../sample_1.000_rep0.002/log.csv")
df['egress_byte'] = df['egress_byte_Presto'] + df['egress_byte_Spark']
mean_egress_size_predict = df[df['mode'] == 'size-predict'].groupby('cloud_computation_target')['egress_byte'].mean()
mean_egress_independent = df[df['mode'] == 'independent'].groupby('cloud_computation_target')['egress_byte'].mean()
mean_egress_size_unaware = df[df['mode'] == 'size-unaware'].groupby('cloud_computation_target')['egress_byte'].mean()
mean_egress_size_aware = df[df['mode'] == 'size-aware'].groupby('cloud_computation_target')['egress_byte'].mean()
# reduction on independent mode
reduction = (mean_egress_independent - mean_egress_size_predict) / mean_egress_independent * 100
print("** size-predict reduction on independent", reduction.round(1))
# reduction on size-unaware mode
reduction = (mean_egress_size_unaware - mean_egress_size_predict) / mean_egress_size_unaware * 100
print("** size-predict reduction on size-unaware", reduction.round(1))
gap = (mean_egress_size_predict - mean_egress_size_aware) / mean_egress_size_aware * 100
print("** size-predict is within ?% of size-aware", gap.round(1))