<jupyter_start><jupyter_text>Scalability and sustainable throughputThis notebook can be used to make scalability and sustainable throughput plots.<jupyter_code># settings for saving plots
saveplots = True
dpiResolution = 100
from pyspark.sql.functions import *
import pandas as pd
import numpy as np
# Import to indent the plots in the notebook
%matplotlib notebook
%matplotlib inline
import matplotlib
import matplotlib.cm as cm
import matplotlib.pyplot as plt
import matplotlib.collections as collections
import seaborn as sns
from IPython.core.display import display, HTML
from PIL import Image
warnings.simplefilter('ignore', Image.DecompressionBombWarning)
import datetime
from matplotlib.lines import Line2D
from matplotlib.ticker import AutoMinorLocator
from pyspark.sql import Window
from pyspark.sql import functions
from pyspark.sql.types import IntegerType, LongType, DoubleType, TimestampType, StringType
# check if Spark is running
plt.style.use('ggplot')
plt.style.use('seaborn-deep')
plt.rcParams['font.family'] = 'sans-serif'
plt.rcParams['font.serif'] = 'Ubuntu'
plt.rcParams['font.monospace'] = 'Ubuntu Mono'
plt.rcParams['font.size'] = 18
plt.rcParams['axes.labelsize'] = 18
plt.rcParams['axes.labelweight'] = 'normal'
plt.rcParams['axes.labelcolor'] = 'black'
plt.rcParams['axes.facecolor'] = 'white'
plt.rcParams['axes.edgecolor'] = 'lightgrey'
plt.rcParams['axes.titlesize'] = 18
plt.rcParams['axes.titleweight'] = 'normal'
plt.rcParams['figure.edgecolor'] = 'lightgrey'
plt.rcParams['xtick.labelsize'] = 18
plt.rcParams['ytick.labelsize'] = 18
plt.rcParams['xtick.color'] = 'black'
plt.rcParams['ytick.color'] = 'black'
plt.rcParams['legend.fontsize'] = 18
plt.rcParams['legend.edgecolor'] = 'white'
plt.rcParams['figure.titlesize'] = 18
plt.rcParams['figure.titleweight'] ='bold'
plt.rcParams['grid.color'] = 'grey'
plt.rcParams['grid.linestyle'] = ':'
plt.rcParams['figure.facecolor'] = 'white'
plt.rcParams['axes.spines.top'] = False
plt.rcParams['axes.spines.right'] = False
plt.rcParams['axes.spines.left'] = False
plt.rcParams['axes.spines.bottom'] = False
plt.rcParams['xtick.direction'] = 'out'
plt.rcParams['ytick.direction'] ='out'
plt.rcParams["date.autoformatter.minute"] = "%H:%M"
plt.rcParams["font.family"] = "Times New Roman"
plt.rc('axes',edgecolor='black')
# check if Spark is running
spark
# List of which frameworks should be included in this comparison
frameworksPrinted = {
    "SPARK": "Spark Streaming",
    "STRUCTUREDSTREAMING": "Structured Streaming",
    "FLINK": "Flink",
    "KAFKASTREAMS": "Kafka Streams"
}
stages=["ingest", "parse", "join", "tumbling \n window", "sliding \n window"]
cluster_scales = {
    "baseline": "5x-4cpu-20gb",
    "horizontal": "10x-4cpu-20gb",
    "vertical": "5x-8cpu-40gb"
}<jupyter_output><empty_output><jupyter_text>General Methods For printing the quantiles of a column:provide a list of quantiles<jupyter_code>def dataRate(factor, last_stage):
    if(int(last_stage)<100):
        return 380*(int(factor)+1)
    else:
        return 190*(int(factor)+1)

dataRateUDF = functions.udf(dataRate, IntegerType())

def factorToTotalCount(factor, last_stage):
    if(int(last_stage)<100):
        return int(380*(int(factor)+1)*30*60/3.1666666667)
    else:
        return 120*30*60*(int(factor)+1)

factorToTotalCountUDF = functions.udf(factorToTotalCount, IntegerType())
def save_img_colored_and_grayscale(path_colored_img):
    if saveplots: 
        plt.savefig(path_colored_img + '.png', dpi=dpiResolution, bbox_inches="tight", pad_inches=0)<jupyter_output><empty_output><jupyter_text>Horizontal scale Read in latency data and transform in the right format for plotting.<jupyter_code>def createFrameworkSustainableBurstPlot(completeResults, last_stage, framework, amt_workers, cpu_reservation, memory_reservation):
    latency_metric = 'p50'
    f, ax = plt.subplots(3, 1, figsize=(2.5, 2.7), sharex=True)
    xlim = 1000000
    if(int(last_stage)<100):
        max_allowed_latency = 10000
    else:
        max_allowed_latency = 90000
        
    print(framework + " - " + amt_workers + "x-" + cpu_reservation +"cpu-" + memory_reservation + "gb")
    data = completeResults\
        .withColumn("scale", col("scale").cast("Integer"))\
        .orderBy('scale')
    
    dfToPlot = data\
        .withColumn("expectedCount", factorToTotalCountUDF(col("scale"), lit(last_stage)))\
        .withColumn("missingCount", col("expectedCount")-col("count"))\
        .withColumn("delayTime", col("max_outputKafkaTimestamp")-col("max_inputKafkaTimestamp"))\
        .withColumn("meanCpuPct", col("mean_cpu")*100/cpu_reservation)\
        .withColumn("p50", col("p50").cast("Integer"))\
        .withColumn("p99", col("p99").cast("Integer"))\
        .withColumn("mean_mem", col("mean_mem").cast("Float"))\
        .withColumn("min_mem", col("min_mem").cast("Float"))\
        .withColumn("max_mem", col("max_mem").cast("Float"))
        
    dfToPlotPandas = dfToPlot.toPandas()
    delayedData = dfToPlot.filter((col("meanCpuPct")>80) | (col("p50")>max_allowed_latency)).toPandas()
    notDelayedData = dfToPlot.filter((col("meanCpuPct")<80) & (col("p50")<max_allowed_latency)).toPandas()
    beginningGreyZone = notDelayedData['scale'].apply(lambda x: dataRate(x, last_stage)).max()
    
    dataSummary = dfToPlot.withColumn("missingPct", col("missingCount")/col("expectedCount"))\
        .withColumn("throughput", dataRateUDF(col("scale"), lit(last_stage)))\
        .select("scale", "throughput", round("missingPct").alias("missingPct"),\
                "delayTime", "p50", "p99", round("meanCpuPct").alias("meanCpuPct"), "mean_mem", "filename")\
        .orderBy("scale")
    dataSummary.show()
    dataSummary.drop("fileName", "mean_mem").show()
    
    ax[0].axvspan(beginningGreyZone, xlim, facecolor='grey', alpha=0.5)
    latency_line = ax[0].plot(notDelayedData['scale'].apply(lambda x: dataRate(x, last_stage)), notDelayedData[latency_metric], marker= 'o', markersize=3.5, label='p50 latency', c='#303030', zorder=2)
    ax[0].set_ylabel('latency', rotation='horizontal', labelpad =40, verticalalignment='center')

    maxLatencyNotDelayed = notDelayedData[latency_metric].max()
    ylimitmax = maxLatencyNotDelayed*1.5 if int(last_stage) < 100 else 60000

    ax[0].set_ylim(ymin=0, ymax=ylimitmax)
    ax[0].set_xlim(xmin=0, xmax=xlim)
    # Make it SECONDS y-axis scale instead of MILLIS
    ax[0].get_yaxis().set_major_formatter(
        matplotlib.ticker.FuncFormatter(lambda x, p: format(int(x/1000), ',').replace(',', ' ') + " s"))
    ax[0].spines['bottom'].set_color('grey')
    ax[0].spines['bottom'].set_visible(True)
    ax[0].spines['left'].set_color('grey')
    ax[0].spines['left'].set_visible(True)
    minor_y_locator_0 = AutoMinorLocator(2)
    ax[0].yaxis.set_minor_locator(minor_y_locator_0)
    ax[0].grid(which='minor', color='grey')
    ax[0].axvline(beginningGreyZone, color='black')

    dataToPlot = dfToPlot.toPandas()
    dataToPlot['scale'] = dataToPlot['scale'].apply(lambda x: dataRate(x, last_stage))
    ax[1].plot(dataToPlot['scale'], dataToPlot['meanCpuPct'], linestyle= '--', marker='o', 
               label='mean CPU', c='#303030', zorder=2, markersize=3.5)
    ax[1].set_ylabel('CPU', rotation='horizontal', horizontalalignment='right', verticalalignment='center')
    ax[1].set_ylim(ymin=0, ymax=120)
    ax[1].axvspan(beginningGreyZone, xlim, facecolor='grey', alpha=0.5)
    ax[1].spines['bottom'].set_color('grey')
    ax[1].spines['bottom'].set_visible(True)
    ax[1].spines['left'].set_color('grey')
    ax[1].spines['left'].set_visible(True)
    minor_y_locator_1 = AutoMinorLocator(2)
    ax[1].yaxis.set_minor_locator(minor_y_locator_1)
    ax[1].get_yaxis().set_major_formatter(
        matplotlib.ticker.FuncFormatter(lambda x, p: str(x).replace('.0', '') + " %"))
    ax[1].grid(which='minor', color='grey')
    ax[1].axvline(beginningGreyZone, color='black')

    ax[2].plot(dataToPlot['scale'], dataToPlot['mean_mem'], linestyle = ":", marker= 'o', markersize=3.5, label='avg heap usage', c='black', zorder=2)            
    ax[2].fill_between(dataToPlot['scale'], dataToPlot['min_mem'], dataToPlot['max_mem'], color='grey', label='heap usage range')
    ax[2].set_ylabel('mem', rotation='horizontal', horizontalalignment='right', verticalalignment='center')
    ax[2].set_yticks([0, 15*1024])
    ax[2].set_xlabel('x100K events/sec')
    ax[2].spines['bottom'].set_color('grey')
    ax[2].spines['bottom'].set_visible(True)
    ax[2].spines['left'].set_color('grey')
    ax[2].spines['left'].set_visible(True)
    ax[2].set_ylim(ymin=0, ymax=30*1024)
    minor_y_locator_2 = AutoMinorLocator(2)
    ax[2].yaxis.set_minor_locator(minor_y_locator_2)
    ax[2].grid(which='minor', color='grey')
    ax[2].axvspan(beginningGreyZone, xlim, facecolor='grey', alpha=0.5, label = 'unsustainable')
    ax[2].axvline(beginningGreyZone, color='black')

    # Make it GB instead of the real MB values on the y axis
    ax[2].get_yaxis().set_major_formatter(
        matplotlib.ticker.FuncFormatter(lambda x, p: format(int(x/1024.0), ',').replace(',', ' ') + " GB"))
    # Make it x1000 seconds instead of the real too large values on the x axis
    ax[2].get_xaxis().set_major_formatter(
        matplotlib.ticker.FuncFormatter(lambda x, p: format(int(x/100000.0), ',').replace(',', ' ')))
    
    ax[2].set_xticks(range(0, xlim, 100000))

    save_img_colored_and_grayscale("./figures/scalability/overall/sustainable_throughput_stage" + last_stage + "_" + framework + "_"  + amt_workers + "x_" + cpu_reservation +"cpu_" + memory_reservation + "g")
    
    dataSummary.coalesce(1).write.mode("overwrite").csv("./figures/scalability/overall/sustainable_throughput_summary_" + framework + "_"  + amt_workers + "x-" + cpu_reservation +"cpu-" + memory_reservation + "gb_scale")
    plt.show()<jupyter_output><empty_output><jupyter_text>Phases that are present in the data<jupyter_code>def getPlotsAndInfoForScale(framework, last_stage, amt_workers, cpu_reservation, memory_reservation):
    dataPath = "./scalability-data/constant-rate/" + framework + "/stage" + last_stage + "/" + amt_workers + "x-" + cpu_reservation +"cpu-" + memory_reservation + "gb/*"
    completeResults = spark.read.option("header", "true").option("inferSchema", "true") \
        .csv(dataPath + "/results.csv/*").withColumn("filename", split(input_file_name(), "/202").getItem(1))
    if completeResults.count()>0:
        volumes = list(completeResults.select("scale").distinct().orderBy("scale").toPandas()["scale"])
        
    containerCheckPhase = spark.read.option("header", "true").option("inferSchema", "true") \
        .csv(dataPath + "/cpu-per-container-timeseries.csv/*")
    amtOfContainers = containerCheckPhase.select("containerName").distinct().count()

    required_amt_containers = {
        "horizontal": 10,
        "vertical": 5,
        "baseline": 5
    }
    
    
    createFrameworkSustainableBurstPlot(completeResults, last_stage, framework, amt_workers, cpu_reservation, memory_reservation)
    
    print(volumes)<jupyter_output><empty_output><jupyter_text>Last stage 3<jupyter_code>framework="STRUCTUREDSTREAMING"
last_stage="3"
getPlotsAndInfoForScale(framework, last_stage, "6", "2", "10")
getPlotsAndInfoForScale(framework, last_stage, "2", "6", "30")
getPlotsAndInfoForScale(framework, last_stage, "6", "6", "30")
getPlotsAndInfoForScale(framework, last_stage, "6", "4", "20")
getPlotsAndInfoForScale(framework, last_stage, "4", "6", "30")
getPlotsAndInfoForScale(framework, last_stage, "6", "1", "5")
getPlotsAndInfoForScale(framework, last_stage, "1", "6", "30")<jupyter_output>STRUCTUREDSTREAMING - 1x-6cpu-30gb
+-----+----------+----------+---------+-----+-----+----------+---------+--------------------+
|scale|throughput|missingPct|delayTime|  p50|  p99|meanCpuPct| mean_mem|            filename|
+-----+----------+----------+---------+-----+-----+----------+---------+--------------------+
|    0|       380|     -14.0|     2598| 2968| 3994|      25.0|1918.7555|0-12-14T10_01_13Z...|
|  200|     76380|       1.0|     2678| 5838| 7695|      74.0| 4658.252|0-12-14T16_09_29Z...|
|  250|     95380|       1.0|    39085|52763|65004|      91.0|6862.5713|0-12-14T11_02_19Z...|
|  500|    190380|       1.0|   264491| null| null|      95.0|6035.8096|0-12-14T10_01_13Z...|
+-----+----------+----------+---------+-----+-----+----------+---------+--------------------+

+-----+----------+----------+---------+-----+-----+----------+
|scale|throughput|missingPct|delayTime|  p50|  p99|meanCpuPct|
+-----+----------+----------+---------+-----+-----+----------+
|    0|       380|     [...]<jupyter_text>Last stage 101<jupyter_code>last_stage="101"
getPlotsAndInfoForScale(framework, last_stage, "6", "1", "5")
getPlotsAndInfoForScale(framework, last_stage, "1", "6", "30")
getPlotsAndInfoForScale(framework, last_stage, "6", "6", "30")
getPlotsAndInfoForScale(framework, last_stage, "6", "4", "20")
getPlotsAndInfoForScale(framework, last_stage, "4", "6", "30")
getPlotsAndInfoForScale(framework, last_stage, "6", "2", "10")
getPlotsAndInfoForScale(framework, last_stage, "2", "6", "30")<jupyter_output>STRUCTUREDSTREAMING - 2x-6cpu-30gb
+-----+----------+----------+---------+------+-------+----------+---------+--------------------+
|scale|throughput|missingPct|delayTime|   p50|    p99|meanCpuPct| mean_mem|            filename|
+-----+----------+----------+---------+------+-------+----------+---------+--------------------+
|    0|       190|       1.0|    43003|  1931|  47944|       2.0|822.08984|0-12-24T19_10_50Z...|
| 1500|    285190|       1.0|    46496| 39964|  64790|      55.0|12711.633|0-12-29T13_11_43Z...|
| 1600|    304190|       1.0|    59490| 44195|  61046|      72.0|12766.231|0-12-29T18_53_54Z...|
| 1700|    323190|       1.0|  1244517|599344| 937398|      52.0| 6501.017|0-12-29T17_15_51Z...|
| 2000|    380190|       1.0|  1584246|737293|1314055|      50.0| 7525.122|0-12-24T19_48_04Z...|
+-----+----------+----------+---------+------+-------+----------+---------+--------------------+

+-----+----------+----------+---------+------+-------+----------+
|scale|throughput|missing[...]
{'fraction_non_alphanumeric': 0.21177218767665348, 'fraction_numerical': 0.05490390050876201, 'mean_word_length': 3.768530997304582, 'pattern_counts': {'":': 11, '<': 29, '<?xml version=': 0, '>': 26, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '591218', 'license': 'Apache-2.0', 'n_tokens_mistral': 5495, 'n_tokens_neox': 4928, 'n_words': 784, 'path': 'result-analysis/Scalability and sustainable throughput.ipynb', 'repo_name': 'bvonheid/open-stream-processing-benchmark', 'size': '14152'}