<jupyter_start><jupyter_text>Parse Json to Pandas DataframesThis script is used to convert json packets into a dictionary where the key is a unique metadata configuration and the value is a Pandas dataframe. The Pandas dataframe has a ds column and a y column corresponding to the timestamp and corresponding value in the time series. The dictionary is then stored in a Pickle file.<jupyter_code>import json
import pandas as pd
import fnmatch
import os
import bz2
import pickle
import gc
# read files in list and convert to pandas dataframes
def load_files(files, file_format):
    dfs = {}
    for file in files:
        # check file format and read appropriately
        if file_format == ".json":
            f = open(file, 'rb')
        else:
            f = bz2.BZ2File(file, 'rb')
        jsons = json.load(f)
        f.close()

        # iterate through packets in file
        for pkt in jsons:
            # create a new dataframe with packet timestamp and values
            df = pd.DataFrame.from_dict(pkt["values"])
            df = df.rename( columns={0:"ds", 1:"y"})
            df["ds"] = pd.to_datetime(df["ds"], unit='s')
            df = df.sort_values(by=["ds"])
            df.y = pd.to_numeric(df['y'], errors='coerce')
            df = df.dropna()
            md = str(pkt["metric"])
            # append generated dataframe and metadata to collection
            try:
                dfs[md] = dfs[md].append(df, ignore_index=True)
            except:
                dfs[md] = df
    return dfs
# take a list of dataframes and their metadata and collapse to a
# collection of unique time series (based on unique metadata)
def collapse_to_unique(dfs_master, dfs_new):
    # iterate through metadata
    dfs_remaining = {}
    for md in dfs_new.keys():
        try:
            # find metadata in our master list
            # if this throws an error, simply add it to the list
            dfs_master[md] = dfs_master[md].append(dfs_new[md], ignore_index=True)
        except:
            dfs_remaining[md] = dfs_new[md]
    return dfs_master, dfs_remaining
# create pickle file containing data
def save_checkpoint(pds, file):
    if file[-4:] != ".pkl":
        file = file + ".pkl"
    f = open(file, "wb")
    pickle.dump(pds, f)
    f.close()
    return file

# load pickle file containing data
def load_checkpoint(file):
    f = open(file, "rb")
    pds = pickle.load(f)
    f.close()
    return pds
# remove all temp pickle files generated during this program
def combine_checkpoints(master_file):
    df = {}
    files = os.listdir()
    for file in files:
        if fnmatch.fnmatch(file, "collapsed_*.pkl"):
            try:
                f = open(file, "rb")
                dfs = pickle.load(f)
                f.close()
                df.update(dfs)
            except:
                continue
            os.system("rm " + file)
        elif fnmatch.fnmatch(file, "raw_*.pkl"):
            os.system("rm " + file)
    f = open(master_file + ".pkl", "wb")
    pickle.dump(df, f)
    f.close()
# load all files and convert to a list of pandas dataframes
def convert_to_pandas(files, file_format, batch_size):
    checkpoints = []
    # # separate files into batches
    batches = [files[batch_size*i:batch_size*(i+1)] for i in range(int(len(files)/batch_size) + 1)]
    print("num_batches", len(batches))
    i = 0
    for batch in batches:
        print(i)
        i += 1
        # get new portion of dataframes and add to master set
        pds_new = load_files(batch, file_format)
        cp = save_checkpoint(pds_new, "raw_" + str(i))
        checkpoints.append(cp)
        gc.collect()

    pds = []
    # iterate checkpoint by checkpoint and add data to unique collection
    # of time series
    collapsed_fs = []
    i = 0
    for cp in checkpoints:
        i += 1
        print(i)
        pds_new = load_checkpoint(cp)
        print(i)
        # load data in batches and combine dataframes
        for f in collapsed_fs:
            pds = load_checkpoint(f)
            pds, pds_new = collapse_to_unique(pds, pds_new)
            save_checkpoint(pds, f)
            gc.collect()
        if len(pds_new) > 0:
            f_new = save_checkpoint(pds_new, "collapsed_" + str(i)) 
            print("Generated ", f_new)
            collapsed_fs.append(f_new)   
        print(i)
        gc.collect()
    return pds
# get all appropriately formatted files in a folder
def retrieve_filenames(path, file_format):
    filenames = []
    for file in os.listdir(path):
            # check if this file has correct ending (regex)
            if fnmatch.fnmatch(file, "*" + file_format):
                f_name = path + file
                filenames.append(f_name)
    return filenames

# get main input arguments and return formatted data
def read_input(data_folder, metric, file_format, batch_size):
    # metric-specific data folder
    folder = data_folder + metric + "/"
    # get all files in folder
    files = os.listdir(folder)

    # automatically detect metric type
    if "quantile" in files:
        metric_type = "summary"
        label = "quantile"
        filenames = retrieve_filenames(folder + "quantile/", file_format)
#         filenames_count = retrieve_filenames(folder + "count/", file_format)
#         filenames_sum = retrieve_filenames(folder + "sum/", file_format)
    elif "bucket" in files:
        metric_type = "histogram"
        label = "le"
        filenames = retrieve_filenames(folder + "bucket/", file_format)
#         filenames_count = retrieve_filenames(folder + "count/", file_format)
#         filenames_sum = retrieve_filenames(folder + "sum/", file_format)
    else:
        metric_type = "counter/gauge"
        label = ""
        filenames = retrieve_filenames(folder, file_format)
    
    pd_frames = convert_to_pandas(filenames, file_format, batch_size)

    return pd_frames
def main():
    print("Formatting Data")
    pd_frames = read_input(input_dir, metric, fformat, batch_size)
    print("Conversion successful")

    master_file = output_dir + metric

    combine_checkpoints(master_file)

    print("Saved data:", master_file)
if __name__ == '__main__':

    # input parameters
    metric = "http_request_duration_microseconds"
    fformat='.json.bz2'
    input_dir = "data/"
    output_dir = ""
    batch_size= 20

    main()<jupyter_output>Formatting Data
num_batches 11
0
1
2
3
4
5
6
7
8
9
10
1
1
Generated  collapsed_1.pkl
1
2
2
Generated  collapsed_2.pkl
2
3
3
Generated  collapsed_3.pkl
3
4
4
Generated  collapsed_4.pkl
4
5
5
Generated  collapsed_5.pkl
5
6
6
Generated  collapsed_6.pkl
6
7
7
Generated  collapsed_7.pkl
7
8
8
Generated  collapsed_8.pkl
8
9
9
Generated  collapsed_9.pkl
9
10
10
Generated  collapsed_10.pkl
10
11
11
11
Conversion successful
Saved data: http_request_duration_microseconds
{'fraction_non_alphanumeric': 0.09236652448889543, 'fraction_numerical': 0.011766436240623622, 'mean_word_length': 2.490759753593429, 'pattern_counts': {'":': 2, '<': 4, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '275951', 'license': 'Apache-2.0', 'n_tokens_mistral': 2196, 'n_tokens_neox': 2150, 'n_words': 643, 'path': 'notebooks/Parse Json to Pandas Dataframes.ipynb', 'repo_name': 'heidsoft/forecast-prometheus', 'size': '6799'}