<jupyter_start><jupyter_text>Notebook for preprocessing data from experiments Set working directory. Default is `experiments_data`.<jupyter_code>working_directory = "2GB-9N-spark-listener"
def get_subdirectories(directory=""):
    subdirectories = []
    p = Path(f"./../{working_directory}/{directory}")
    for item in p.glob('*/'):
        if item.suffix not in (['.csv', '.zip']):
            subdirectories.append(directory + "/" + item.name)
    return subdirectories

def get_timestamp_info(data):
    data['timestamp'] = pd.to_datetime(data['timestamp'])
    start_time = data['timestamp'].min()
    stop_time = data['timestamp'].max()

    data['timestamp'] = pd.to_numeric(data['timestamp'])
    mean_interval = mean(data.diff(axis=0)['timestamp'][1:].tolist())
    mean_interval = round(mean_interval / 1000000000, 3)

    return (stop_time - start_time).total_seconds(), mean_interval

def preprocess_file(file_path, save_path, filename, iteration):
    read_data = pd.read_csv(file_path + "/" + filename)
    agg_read_data = read_data.\
        groupby('timestamp', as_index=False).\
        agg({"CPU": "sum", "RAM": "sum"})

    generate_plot(agg_read_data.index, agg_read_data.CPU, "{0}/CPU/{1}_CPU.png".format(save_path, iteration), ylabel="Procentowe zużycie CPU")
    generate_plot(agg_read_data.index, agg_read_data.RAM, "{0}/RAM/{1}_RAM.png".format(save_path, iteration), ylabel="Procentowe zużycie RAM")

    return agg_read_data, get_timestamp_info(agg_read_data)

def generate_plot(data_x="", data_y="", plot_path="", title="", ylabel=""):
    fig = plt.figure(figsize=(10,10))
    fig.patch.set_facecolor('white')
    plt.plot(data_x, data_y, 'r.-')
    plt.title(title)
    plt.xlabel("Numer próbki", fontsize=15)
    plt.ylabel(ylabel, fontsize=15)
    plt.savefig(plot_path)
    plt.close(fig)

def create_directory(path):
    try:
        os.mkdir(path)
    except OSError as e:
        print("Error: %s - %s." % (e.filename, e.strerror))<jupyter_output><empty_output><jupyter_text>Check for subdirectories. Number of subdirectories should much the number of nodes used for experiments.<jupyter_code>all_directories = get_subdirectories()
nodes_directories = [x for x in all_directories if "node" in x]
nodes_directories
data_directories = []
data_directories_groups = []
for directory in nodes_directories:
    cur_node_subdirectories = get_subdirectories(directory)
    data_directories.append(cur_node_subdirectories)

data_directories_groups = data_directories
data_directories = [item for sublist in data_directories for item in sublist]
data_directories_groups<jupyter_output><empty_output><jupyter_text>Check for function names in each node directory.These names should match the names in `experiments-plan.csv`.<jupyter_code>function_names = data_directories_groups[1]
function_names = list(map(lambda x: x[8:], data_directories_groups[1]))
function_names<jupyter_output><empty_output><jupyter_text>Aggregating data from all nodes (11-19)<jupyter_code>master_dir_index = nodes_directories.index('/node_20')
workers_directories = nodes_directories.copy()
del workers_directories[master_dir_index]

workers_directories
create_directory(f"./../{working_directory}/preprocessed-data")

mean_dir = f"./../{working_directory}/preprocessed-data/workers-mean-data"
create_directory(mean_dir)
workers_summary = pd.DataFrame(columns=["function_name",
                                             "mean_duration",
                                             "mean_interval",
                                             "nodes_durations",
                                             "nodes_intervals"])

workers_summary.to_csv("{0}/experiments_mean_duration.csv".format(mean_dir), index=False)
def preprocess_selected_function(f_names, workers_summary):
    for function_name in f_names:
        base_data = pd.DataFrame()
        experiment_duration = []
        experiment_interval = []
        nodes_durations = []
        nodes_intervals = []

        experiment_mean_dir = "{0}{1}".format(mean_dir, function_name)
        create_directory(experiment_mean_dir)

        for node_dir in workers_directories:
            file_path = f"./../{working_directory}{node_dir}{function_name}"
            p = Path(file_path)
            plots_path = "{0}/plots".format(file_path)

            create_directory(plots_path)
            create_directory(plots_path + "/RAM")
            create_directory(plots_path + "/CPU")

            experiment_number = 1
            node_intervals = []
            node_durations = []

            for file in p.glob('*.csv'):
                new_data, (duration, interval) = preprocess_file(file_path, plots_path, file.name, experiment_number)
                base_data = pd.concat((base_data, new_data))

                experiment_duration.append(duration)
                experiment_interval.append(interval)
                node_intervals.append(interval)
                node_durations.append(duration)

                experiment_number += 1

            print("{0} | {1} durations: {2}".format(function_name[1:], node_dir[1:], node_durations))

            nodes_intervals.append((node_dir[1:], round(mean(node_intervals), 3)))
            nodes_durations.append((node_dir[1:], round(mean(node_durations), 3)))

        mean_interval = round(mean(experiment_interval), 3)

        base_data["timestamp"] = base_data.apply(lambda row: mean_interval * row.name, axis = 1)
        base_data = base_data.groupby(base_data.index).mean()
        base_data.to_csv("{0}/mean_data.csv".format(experiment_mean_dir), index=False)

        generate_plot(base_data.index, base_data.CPU, "{0}/mean_CPU.png".format(experiment_mean_dir), function_name[1:] + " CPU", ylabel="Procentowe zużycie CPU")
        generate_plot(base_data.index, base_data.RAM, "{0}/mean_RAM.png".format(experiment_mean_dir), function_name[1:] + " RAM", ylabel="Procentowe zużycie RAM")

        workers_summary = workers_summary.append({"function_name": function_name[1:],
                                                            "mean_duration": round(mean(experiment_duration), 3),
                                                            "mean_interval": mean_interval,
                                                            "nodes_durations": nodes_durations,
                                                            "nodes_intervals": nodes_intervals}, ignore_index=True)

    workers_summary.to_csv("{0}/experiments_mean_duration.csv".format(mean_dir), index=False, mode="a", header=False)

    del base_data
    del experiment_interval
    del experiment_duration
    del nodes_durations
    del nodes_intervals
    del new_data
    del node_intervals
    del node_durations
    del workers_summary
    gc.collect()
print(f"Functions: {len(function_names)}")
preprocess_selected_function(function_names[0:4], workers_summary)
gc.collect()
preprocess_selected_function(function_names[4:8], workers_summary)
gc.collect()
preprocess_selected_function(function_names[8:12], workers_summary)
gc.collect()
preprocess_selected_function(function_names[12:16], workers_summary)
gc.collect()
preprocess_selected_function(function_names[16:20], workers_summary)
gc.collect()
preprocess_selected_function(function_names[20:24], workers_summary)
gc.collect()
preprocess_selected_function(function_names[24:28], workers_summary)
gc.collect()
preprocess_selected_function(function_names[28:], workers_summary)
gc.collect()<jupyter_output><empty_output><jupyter_text>Aggregating data from master node 20<jupyter_code>create_directory(f"./../{working_directory}/preprocessed-data")

master_dir = f"./../{working_directory}/preprocessed-data/master-mean-data"
create_directory(master_dir)
master_mean_summary = pd.DataFrame(columns=["function_name",
                                            "mean_duration",
                                            "mean_interval"])

print("Progress:")
for function in function_names:
    files_path = f"./../{working_directory}/node_20{function}"
    p = Path(files_path)
    master_data = pd.DataFrame()

    experiment_durations = []
    experiment_intervals = []

    master_experiment_directory = "{0}{1}".format(master_dir, function)
    master_plot_directory = "{0}/plots".format(files_path)

    create_directory(master_experiment_directory)
    create_directory(master_plot_directory)
    create_directory("{0}/CPU".format(master_plot_directory))
    create_directory("{0}/RAM".format(master_plot_directory))

    experiment_number = 1
    for file in p.glob('*.csv'):
        master_new_data, (duration, interval) = preprocess_file(files_path, master_plot_directory, file.name, experiment_number)
        master_data = pd.concat((master_data, master_new_data))
        experiment_durations.append(duration)
        experiment_intervals.append(interval)
        experiment_number += 1

    master_mean_summary = master_mean_summary.append({"function_name": function[1:],
                                                      "mean_duration": round(mean(experiment_durations), 3),
                                                      "mean_interval": round(mean(experiment_intervals), 3)},
                                                     ignore_index=True)

    master_data = master_data.groupby(master_data.index).mean()
    master_data.to_csv("{0}/mean_data.csv".format(master_experiment_directory), index=False)

    generate_plot(master_data.index, master_data.CPU, "{0}/mean_CPU.png".format(master_experiment_directory), ylabel="Średnie procentowe zużycie CPU")
    generate_plot(master_data.index, master_data.RAM, "{0}/mean_RAM.png".format(master_experiment_directory), ylabel="Średnie procentowe zużycie RAM")

    print("- {0} - done".format(function[1:]))

    # Deleting variables
    del master_data
    del experiment_intervals
    del experiment_durations
    del master_new_data

master_mean_summary.to_csv("{0}/experiments_mean_duration.csv".format(master_dir), index=False)

del master_mean_summary
gc.collect()<jupyter_output>Error: ./../2GB-9N-spark-listener/preprocessed-data - Nie można utworzyć pliku, który już istnieje.
Progress:<jupyter_text>Preprocessing spark internal metricsStage and task metrics<jupyter_code>stage_metrics_all = pd.read_csv(f"./../{working_directory}/stage_metrics.csv")
task_metrics_all = pd.read_csv(f"./../{working_directory}/task_metrics.csv")

stage_metrics = stage_metrics_all[["function_name", "stage_id", "num_tasks", "executor_run_time", "result_size"]].copy()
stage_metrics["stage_time"]= stage_metrics_all["completion_time"] - stage_metrics_all["submission_time"]
stage_metrics = stage_metrics.groupby(["function_name", "stage_id"]).agg({
    "stage_time": "mean",
    "num_tasks": "max",
    "executor_run_time": "mean",
    "result_size": "mean"
})

task_metrics = task_metrics_all[["function_name", "stage_id", "task_type"]]
task_metrics = task_metrics.groupby(["function_name", "stage_id"]).agg({"task_type": "unique"})

spark_metrics = task_metrics.join(stage_metrics).reset_index().rename(columns={
    "stage_time": "mean_stage_time",
    "task_type": "task_types",
    "executor_run_time": "mean_executor_run_time",
    "result_size": "mean_result_size"
})

spark_metrics.to_csv(f"./../{working_directory}/preprocessed-data/spark_metrics.csv", index=False)
spark_metrics<jupyter_output><empty_output><jupyter_text>Clear preprocessed data<jupyter_code>for function_name in function_names:
    for node_dir in nodes_directories:
        try:
            shutil.rmtree(f"./../{working_directory}{node_dir}{function_name}/plots")
        except OSError as e:
            print("Error: %s - %s." % (e.filename, e.strerror))

try:
    shutil.rmtree(f"./../{working_directory}/preprocessed-data")
except OSError as e:
    print("Error: %s - %s." % (e.filename, e.strerror))<jupyter_output><empty_output><jupyter_text>Clear smoothed data<jupyter_code>for function_name in function_names:
    try:
        os.remove(f'./../experiments_data/preprocessed-data/workers-mean-data{function_name}/smooth_mean_CPU.png')
        os.remove(f'./../experiments_data/preprocessed-data/workers-mean-data{function_name}/smooth_mean_data.csv')
        os.remove(f'./../experiments_data/preprocessed-data/workers-mean-data{function_name}/smooth_mean_RAM.png')
    except OSError as e:
        print("Error: %s - %s." % (e.filename, e.strerror))<jupyter_output><empty_output><jupyter_text>Plots with 150 samples<jupyter_code>preprocess_file("./../1GB-5N/node_11/minNetProfitGroupedBySoldDateWhereYearAfter2000", "./../overleaf/figures/first", "08_08_2021_09_56_17.csv", 5)
preprocess_file("./../1GB-5N/node_11/filterCatalogSalesWhereProfitNegative", "./../overleaf/figures/first", "08_08_2021_02_06_21.csv", 17)
preprocess_file("./../1GB-5N/node_11/avgWholeSaleCostGroupedBySoldDate", "./../overleaf/figures/first", "07_08_2021_11_34_29.csv", 16)
preprocess_file("./../1GB-5N/node_11/maxNetProfitGroupedBySoldDateWhereProfitNegative", "./../overleaf/figures/first", "08_08_2021_07_04_51.csv", 2)
preprocess_file("./../1GB-5N/node_11/filterStoreSalesWhereYearAfter2000", "./../overleaf/figures/first", "08_08_2021_03_35_44.csv", 12)<jupyter_output><empty_output><jupyter_text>Overhead measurement<jupyter_code>data_listener = pd.read_csv(f"./../1GB-9N-spark-listener/preprocessed-data/workers-mean-data/experiments_mean_duration.csv")[["function_name","mean_duration"]]
data_no_listener = pd.read_csv(f"./../1GB-9N/preprocessed-data/workers-mean-data/experiments_mean_duration.csv")[["function_name","mean_duration"]]
listener_mean_data_comparison = pd.merge(data_listener, data_no_listener, on="function_name")
listener_mean_data_comparison["delta_length [listener minus no listener]"] = round(listener_mean_data_comparison["mean_duration_x"] - listener_mean_data_comparison["mean_duration_y"], 3)
listener_mean_data_comparison = listener_mean_data_comparison.rename(columns={"mean_duration_x": "mean_duration_listener",
                                              "mean_duration_y": "mean_duration_no_listener"})
listener_mean_data_comparison.to_csv(f"./../1GB-9N-spark-listener/preprocessed-data/workers-mean-data/mean_duration_comparison.csv", index=False)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12803376714737952, 'fraction_numerical': 0.01575800211044671, 'mean_word_length': 4.053679345894063, 'pattern_counts': {'":': 21, '<': 40, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '270092', 'license': 'Apache-2.0', 'n_tokens_mistral': 4621, 'n_tokens_neox': 4427, 'n_words': 623, 'path': 'notebooks/prepare_data.ipynb', 'repo_name': 'pientaa/opening-black-box', 'size': '14215'}