Migrate Beam benchmark implementation

Co-authored-by: Jan Bensien stu128012@mail.uni-kiel.de

This merge request will fix #147 (closed). It is based on the existing implementation in the https://git.se.informatik.uni-kiel.de/stu128012/theodolite/-/tree/beam_current_theodolite fork but has to tackle the following requirements:

• Migrate the individual benchmarks with their new enumeration
  • Migrate UC1
  • Migrate UC2 (f.k.a. UC3)
  • Migrate UC3 (f.k.a. UC4)
  • Migrate UC4 (f.k.a. UC2)
• Migrate to Titan Control Center Commons 0.1.0 (Avro records and Schema Registry are already used)
• Fix code quality issues if any
• Support Docker builds (already available in the base branch)
• Build and push Docker images in CI pipeline
• Update Docker Compose files in docker-test
• Share code among Beam runners (either create subprojects or dedicated Gradle tasks or dedicated Dockerfiles or environment variables or ...)
  • Current approach is to use different subprojects. Other approaches would increase the complexity and is currently not possible because of:
    • https://issues.apache.org/jira/browse/BEAM-9830
    • https://stackoverflow.com/questions/61441333/conflict-with-runner-dependencies-in-beam
    • build beam commons project (for builing configurations, common PT transformations)
    • outsource uc1-4 pipelines into single projects
    • outsource uc1-4 beam samza into single projects
    • outsource uc1-4 beam flink into single projects
• Clean up Gradle build files
  • Build Gradle plugins:
    • Build beam commons plugin
    • Build beam flink plugin
    • Build beam samza plugin
  • Check if same versions of dependecies are used (slf4, etc)
• Provide corresponding Kubernetes benchmark definitions -> Moved to #281 (closed)
• Solve #278 (closed)
• Fix #293 (closed)
• Solve #291 (not part of this MR)
• Make samza configurable through environment variables
• Resolve ./gradlew build --warning-mode all
• @she Image names in Docker Compose files must be replaced when merging and pushing images

Depends on !188 (merged)

analysis/src/demand.py

 import os
 from datetime import datetime, timedelta, timezone
 import pandas as pd
+from pandas.core.frame import DataFrame
 from sklearn.linear_model import LinearRegression
 
 def demand(exp_id, directory, threshold, warmup_sec):
     raw_runs = []
 
-    # Compute SL, i.e., lag trend, for each tested configuration
-    filenames = [filename for filename in os.listdir(directory) if filename.startswith(f"exp{exp_id}") and filename.endswith("totallag.csv")]
+    # Compute SLI, i.e., lag trend, for each tested configuration
+    filenames = [filename for filename in os.listdir(directory) if filename.startswith(f"exp{exp_id}") and "lag-trend" in filename and filename.endswith(".csv")]
     for filename in filenames:
-        #print(filename)
         run_params = filename[:-4].split("_")
-        dim_value = run_params[2]
-        instances = run_params[3]
+        dim_value = run_params[1]
+        instances = run_params[2]
 
         df = pd.read_csv(os.path.join(directory, filename))
-        #input = df.loc[df['topic'] == "input"]
         input = df
-        #print(input)
+
         input['sec_start'] = input.loc[0:, 'timestamp'] - input.iloc[0]['timestamp']
-        #print(input)
-        #print(input.iloc[0, 'timestamp'])
+    
         regress = input.loc[input['sec_start'] >= warmup_sec] # Warm-Up
-        #regress = input
 
-        #input.plot(kind='line',x='timestamp',y='value',color='red')
-        #plt.show()
+        X = regress.iloc[:, 1].values.reshape(-1, 1)  # values converts it into a numpy array
+        Y = regress.iloc[:, 2].values.reshape(-1, 1)  # -1 means that calculate the dimension of rows, but have 1 column
 
-        X = regress.iloc[:, 2].values.reshape(-1, 1)  # values converts it into a numpy array
-        Y = regress.iloc[:, 3].values.reshape(-1, 1)  # -1 means that calculate the dimension of rows, but have 1 column
         linear_regressor = LinearRegression()  # create object for the class
         linear_regressor.fit(X, Y)  # perform linear regression
         Y_pred = linear_regressor.predict(X)  # make predictions
 
         trend_slope = linear_regressor.coef_[0][0]
-        #print(linear_regressor.coef_)
 
         row = {'load': int(dim_value), 'resources': int(instances), 'trend_slope': trend_slope}
-        #print(row)
         raw_runs.append(row)
 
     runs = pd.DataFrame(raw_runs)
 
-    # Set suitable = True if SLOs are met, i.e., lag trend is below threshold
-    runs["suitable"] =  runs.apply(lambda row: row['trend_slope'] < threshold, axis=1)
-
-    # Sort results table (unsure if required)
-    runs.columns = runs.columns.str.strip()
-    runs.sort_values(by=["load", "resources"])
+    # Group by the load and resources to handle repetitions, and take from the reptitions the median
+    # for even reptitions, the mean of the two middle values is used
+    medians = runs.groupby(by=['load', 'resources'], as_index=False).median()
 
-    # Filter only suitable configurations
-    filtered = runs[runs.apply(lambda x: x['suitable'], axis=1)]
-
-    # Compute demand per load intensity
-    grouped = filtered.groupby(['load'])['resources'].min()
-    demand_per_load = grouped.to_frame().reset_index()
+    # Set suitable = True if SLOs are met, i.e., lag trend slope is below threshold
+    medians["suitable"] =  medians.apply(lambda row: row['trend_slope'] < threshold, axis=1)
 
+    suitable = medians[medians.apply(lambda x: x['suitable'], axis=1)]
+    
+    # Compute minimal demand per load intensity
+    demand_per_load = suitable.groupby(by=['load'], as_index=False)['resources'].min()
+    
     return demand_per_load
+