# # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. # # Experiments consist of a # - workload # - data generator # - query generator # - oracle # - sketches # # Running an experiment will generate a csv containing the experimental results # This can be fed into a function/script/notebook to process and plot the results # # For an existing problem and new, experimental sketch, one just needs to create a sketch that wraps the # implementation with the simple API used in the test. # # For a novel problem, one must also write a query generator and an oracle that will compute the correct answer and calculate the error # when comparing to a sketch's answer. # # # Replicates in an experiment come from # - reordering / generating new data sequences (runs) # - randomization at the sketch level (repetitions) # To make things parallelizable we # - For each run, write a workload's data sequence to disk (unless it's just drawing from a distribution) # - Compute the oracle's answers for that sequence and cache it on disk # - Create a bunch of jobs to evaluate a sketch on the data sequence using the cached answers # # Since our jobs contain nested class instances, python's default pickle doesn't work # So we use dill to pickle the instances ourselves and unpickle when its run ############################################################################################################################# import os, psutil import logging import time from random import uniform import pandas as pd import numpy as np from enum import Enum import copy import dill # needed to pickle classes/functions for use in multiprocessing from collections import deque import importlib from concurrent.futures import ProcessPoolExecutor from experiment_utils import makeDict ############ class ResultAccumulator: def __init__(self, save_answers=False): self.results = [] self.timings = [] self.save_answers = save_answers def addResult(self, sketch, error, answer, truth, query, workload): if not self.save_answers: answer = None query_info = query.info() sketch_info = sketch.info() workload_info = workload.info() result = makeDict(error=error, answer=answer, truth=truth, **query_info, **workload_info, **sketch_info) #qid=query.qid, **workload_info, **sketch_info) self.results.append(result) def extend(self, results: list): self.results.extend(results) def merge(self, results): self.results.extend(results.results) self.timings.extend(results.timings) def toDataFrame(self): df = pd.DataFrame(self.results) return df def addTiming(self, run_time, sketch, workload): sketch_info = sketch.info() workload_info = workload.info() time_result = makeDict(time=run_time, **workload_info, **sketch_info) self.timings.append(time_result) def timingsToDataFrame(self): df = pd.DataFrame(self.timings) return df class ExperimentOptions: class OracleMode(Enum): PRECOMPUTE = 0 # standard case which can be serialized to a C++/Java test POSTCOMPUTE = 1 # for when the oracle needs a large datastructure to evaluate the error JOINTLY_COMPUTE = 2 # likely to lead to bad timings and memory def __init__( self, ndatasets=1, # number of times to generate or reorder the dataset nrepetitions=10, # number of times to repeat a given ordering. (so the true answers do not need to be recomputed) oracle_mode=OracleMode.PRECOMPUTE, nparallel=1, # number of parallel processes maxtasks=1, # max number of concurrent workload instances log_file=None, save_answers=False, ): self.ndatasets = ndatasets self.nrepetitions = nrepetitions self.oracle_mode = oracle_mode self.nparallel = nparallel self.maxtasks = maxtasks self.log_file = log_file self.save_answers = save_answers #################################################################### # # Classes used to run parallel processes # # Task = Each data stream in the experiment + the oracle + all the sketches # Job = every pass through the stream. Each job has its sketch randomized with a different seed # # The basic architecture is that an experiment generates Tasks. # Each Task generates a collection of Jobs # If the data streams are processed in parallel # Each Job uses dill to pickle the Job # Each child process unpickles a Job and runs it # # Note that some work may be required to ensure all relevant variables are pickled or to ensure # the enclosing environment for the unpickled job matches the original environment # #################################################################### class Job: """ given an instantiated workload, evaluates the specified sketch """ # assumes workload and oracle are already appropriately initialized def __init__(self, repetition, sketch_name, sketch_fun, sketch_kwargs, workload, oracle, options): self.repetition = repetition self.sketch_name = sketch_name self.sketch_fun = sketch_fun self.sketch_kwargs = sketch_kwargs self.sketch_name = sketch_name self.workload = workload self.oracle = oracle self.results = ResultAccumulator(save_answers=True) # XXX self.options = options def executeJob(job): job.workload.prepare() job.oracle.prepare() print(f"Running {job.workload.getID()}, repetition {job.repetition}, on sketch {job.sketch_name} {str(job.sketch_kwargs)} ") query_iter = job.workload.genQueries() # XXX just workload q = next(query_iter) sketch = job.sketch_fun(seed=job.repetition, **job.sketch_kwargs) sketch.name = job.sketch_name results = ResultAccumulator(save_answers=job.options.save_answers) start_time = time.perf_counter() for i, x in enumerate(job.workload.genData()): sketch.add(x) while q and i == q.data_idx: answer = sketch.query(q.query, q.parameters) error = job.oracle.eval_sketch_answer(q.qid, answer) truth = job.oracle.getAnswer(q.qid) results.addResult( error=error, answer=answer, truth=truth, query=q, sketch=sketch, workload=job.workload) q = next(query_iter, None) run_time = time.perf_counter() - start_time results.addTiming(run_time=run_time, sketch=sketch, workload=job.workload) return results def executePickledJob(pickled_job): job = dill.loads(pickled_job) return executeJob(job) class Task: """ controls multiple runs of a workload """ def __init__(self, run, sketch_seed_start, experiment, executor=None): self.run = run self.workload = copy.deepcopy(experiment.workload) self.oracle = copy.deepcopy(experiment.oracle) self.experiment = experiment self.executor = executor self.sketch_seed_start = sketch_seed_start # create jobs for each process # Currently does not parallelize oracle preparation def makeJobs(self): print(f"Making jobs for run {self.run}") self.workload.reset(seed=self.run) self.oracle.reset(workload=self.workload) self.oracle.prepare() workload = self.workload oracle = self.oracle workload.prepareForPickle() oracle.prepareForPickle() for rep in range(self.experiment.options.nrepetitions): for sketch_name, (sketch_fun, sketch_params) in self.experiment.sketches.items(): for p in sketch_params: yield Job(self.sketch_seed_start + rep, sketch_name, sketch_fun, p, workload, oracle, options=self.experiment.options) #################################################################################################### class SketchExperiment: """ Run an experiment for given workload over a collection of sketches The workloads can be randomized with a seed or they can be repeated for randomized sketches """ name = "BaseExperiment" target_results = None # iterable providing the corresponding results (or a way to compute the result) sketches = None # dictionary of sketches MAX_SKETCHES = 100000 def __init__(self, workload, oracle, options=None, result_file="exp_results.csv"): # basic setup variables self.workload = workload self.oracle = oracle self.options = options if options is not None else ExperimentOptions() self.results = ResultAccumulator() self.timer = {} self.result_file=result_file self.start_seed = 0 def getNumSketches(self): num = 1 for v in self.experiment.sketches.values(): num *= len(v) return num def addSketches(self, sketches): self.sketches = sketches def setOptions(self, options): self.options = options def prepareOracle(self, size=None, **kwargs): raise Exception def prepare(self, **kwargs): logging.info("prep workload0") self.workload.prepare() def execute(self, write_mode="w"): assert(self.options.oracle_mode == self.options.OracleMode.PRECOMPUTE) if self.options.nparallel > 1: executor = ProcessPoolExecutor(max_workers=self.options.nparallel) for run in range(self.options.ndatasets): logging.info(f"Starting run {run}") workload_seed = self.start_seed + run sketch_seed_start = workload_seed * self.MAX_SKETCHES task = Task(workload_seed, sketch_seed_start, experiment=self) futures = deque() job_sizes = [] for job in task.makeJobs(): if self.options.nparallel > 1: pickled_job = dill.dumps(job) job_sizes.append(len(pickled_job)) futures.append(executor.submit(executePickledJob, pickled_job)) else: job_result = executeJob(job) self.results.merge(job_result) job_sizes = np.array(job_sizes) print(f"pickled job max size: {max(job_sizes)} avg size: {np.mean(job_sizes)}") if self.options.nparallel > 1: # block when there are enough parallel jobs running while futures and len(futures) >= self.options.nparallel: f = futures.popleft() job_result = f.result() self.results.merge(job_result) # cleanup last jobs while futures: f = futures.popleft() job_result = f.result() self.results.merge(job_result) print(f"writing output {self.result_file} mode: {write_mode}") self.results.toDataFrame().to_csv(self.result_file, mode=write_mode, header=(write_mode == 'w')) timing_df = self.results.timingsToDataFrame() agg_timing_df = timing_df.groupby(['sketch_name']).agg(runtime=('time', np.mean)) print(agg_timing_df.to_string()) # TODO: clean up the temporary answer and pickle files #################################################################################################### class SketchMetaExperiment: """ Run experiments on multiple workloads """ def __init__(self, workloads, oracle, options=None, result_file="exp_results.csv"): self.workloads = workloads self.oracle = oracle self.options = options self.result_file=result_file self.sketches = [] def execute(self, default_write_mode='w'): write_mode = default_write_mode for w in self.workloads: self.oracle.reset(workload=w) e = SketchExperiment(workload=w, oracle=self.oracle, options=self.options, result_file=self.result_file) e.addSketches(self.sketches) e.prepare() e.execute(write_mode=write_mode) write_mode = "a" def addSketches(self, sketches): self.sketches = sketches