| #!/usr/bin/env python |
| |
| # 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. |
| |
| import sys |
| import argparse |
| from datetime import datetime |
| import pandas as pd |
| import numpy as np |
| from utils import distinct_number_sequence |
| import datasketches as ds |
| import datasketch as d |
| import mmh3 |
| import os |
| import matplotlib.pyplot as plt |
| |
| class ErrorCardinalityProfile: |
| """Generates an experiment comparing the error over different cardinalities""" |
| def __init__(self, sketch_lgk:int, lg_trials:int, max_lgN:int): |
| self.sketch_lgk = sketch_lgk |
| self.num_trials = 2**lg_trials |
| self.max_lgN = max_lgN |
| self.max_num_distincts = np.uint64(2 ** self.max_lgN) |
| self.directory_name = "hll_accuracy_profile_" + datetime.today().strftime('%Y%m%d') |
| if not os.path.exists(self.directory_name): |
| os.mkdir(self.directory_name) |
| self.file_extension = "_" + datetime.today().strftime('%H%M') + f"lgK_{self.sketch_lgk}_lgT_{lg_trials}" |
| |
| # Need to remove repeated items for the program logic in self.run() |
| self.plot_points = self._generate_plot_points() |
| self.plot_points.extend(self._generate_plot_points()) |
| self.plot_points = list(set(self.plot_points)) |
| self.plot_points.sort() |
| print(self.plot_points) |
| |
| # Initialise the data structures for results |
| self.DataSketches_results_arr = np.zeros((len(self.plot_points), self.num_trials), dtype=float) |
| self.datasketch_results_arr = np.zeros_like(self.DataSketches_results_arr) |
| self.DataSketches_results_df = pd.DataFrame(index=self.plot_points, columns=None) |
| self.datasketch_results_df = pd.DataFrame(index=self.plot_points, columns=None) |
| self.data = np.random.randn(len(self.plot_points), self.num_trials) |
| print("Data shape: ", self.data.shape) |
| |
| def _generate_plot_points(self) -> list: |
| """ |
| Generates the standard sequence defining the input cardinalites for the experiment |
| This is just two points at each power of 2 |
| """ |
| all_plot_points = [] |
| for lgk in range(1, self.max_lgN+1): |
| points = np.unique(np.logspace(start=lgk, stop=lgk+1, num=8, endpoint=False, base=2, dtype=np.uint64)) |
| all_plot_points.extend(points) |
| all_plot_points.sort() |
| return all_plot_points |
| |
| def _bad_hll_range(self) -> list: |
| """Generates 16 logspaced points in the n ≈ 2.5k range.""" |
| log2_sketch_threshold = np.log2(2.5* (2**self.sketch_lgk)) |
| start = np.floor(log2_sketch_threshold).astype(np.uint64) |
| stop = np.ceil(log2_sketch_threshold).astype(np.uint64)+1 |
| points = np.logspace(start, stop, num=50, endpoint=False, base=2, dtype=np.uint64)[1:] |
| return points |
| |
| def _is_power_of_two(self, a): |
| """Bitwise operations to check value a is a power of two""" |
| return (a & (a-1) == 0) and a != 0 |
| |
| def _results_to_df(self, start_:int, end_:int, arr:np.array, df:pd.DataFrame): |
| """Concatenates the array between columns start_,...end_ - 1 to the dataframe""" |
| new_df = pd.DataFrame(arr[:, start_:end_], index=df.index, columns=np.arange(start_, end_).tolist()) |
| print("concatenating: ", new_df) |
| concat_df = pd.concat([df, new_df], axis=1) |
| return concat_df |
| |
| def run(self): |
| """Runs the experiment""" |
| seq_start = np.uint64(2345234) |
| distinct_number = np.uint64(3462) |
| previous_log_trial_index = 0 |
| ds_all_results = np.zeros((self.num_trials, len(self.plot_points))) |
| d_all_results = np.zeros_like(ds_all_results) |
| |
| for trial in range(1, self.num_trials+1): |
| #print(f"Trial = {trial}\t{self._is_power_of_two(trial)}") |
| # Initialise the sketches |
| hll = ds.hll_sketch(self.sketch_lgk, ds.HLL_8) |
| h = d.HyperLogLogPlusPlus(p=self.sketch_lgk, hashfunc=lambda x: mmh3.hash64(x, signed=False)[0]) |
| plot_point_index = 0 # Return to the start of the plot points list to generate the data |
| plot_point_value = self.plot_points[plot_point_index] |
| total_updates = 0 |
| seq_start += distinct_number # Start a new input sequence |
| |
| # Temporary result data structure |
| ds_results = np.zeros((len(self.plot_points),)) |
| d_results = np.zeros_like(ds_results) |
| |
| |
| for new_number in distinct_number_sequence(seq_start): |
| hll.update(new_number) |
| h.update(str(new_number).encode('utf8')) |
| total_updates += 1 |
| #print(f"Trial: {trial:<5} updates: {total_updates:<5}Index: {plot_point_index:<5} Value: {plot_point_value:<5}\n") |
| if total_updates == plot_point_value: |
| ds_results[plot_point_index] = (hll.get_estimate() - plot_point_value) / plot_point_value |
| d_results[plot_point_index] = (h.count() - plot_point_value) / plot_point_value |
| plot_point_index += 1 |
| #print(f"PPI:{plot_point_index:<3}PPV:{plot_point_value}") |
| |
| if plot_point_index < len(self.plot_points): |
| plot_point_value = self.plot_points[plot_point_index] |
| else: |
| #print("Already at the end") |
| break |
| |
| # After the break statement, control returns here. Now need to decide whether to write or continue. |
| ds_all_results[trial-1, :] = ds_results # subtract 1 as we use 1-based indexing for the trial count. |
| d_all_results[trial - 1, :] = d_results # subtract 1 as we use 1-based indexing for the trial count. |
| if self._is_power_of_two(trial) and trial > 1: |
| # write the array only a logarithmic number of times |
| temporary_ds_results = ds_all_results[0:trial, : ] |
| temporary_d_results = d_all_results[0:trial, :] |
| print(f"#################### PARTIAL RESULTS FOR {trial} TRIALS: DATASKETCHES ####################") |
| previous_log_trial_index = trial |
| self.DataSketches_results_df = pd.DataFrame(temporary_ds_results.T, index=self.DataSketches_results_df.index, columns=np.arange(trial).tolist()) |
| self.DataSketches_results_df.to_csv( |
| self.directory_name + "/DataSketches_hll" + self.file_extension + f"trials_{trial}.csv", |
| index_label="n") |
| self.datasketch_results_df = pd.DataFrame(temporary_d_results.T, |
| index=self.datasketch_results_df.index, |
| columns=np.arange(trial).tolist()) |
| self.datasketch_results_df.to_csv( |
| self.directory_name + "/datasketch_hll" + self.file_extension + f"trials_{trial}.csv", |
| index_label="n" |
| ) |
| print(self.DataSketches_results_df) |
| |
| print(f"#################### PARTIAL RESULTS FOR {trial} TRIALS: datasketch ####################") |
| print(self.datasketch_results_df) |
| |
| |
| def main(argv): |
| assert len(argv) == 3 |
| SKETCH_LGK = argv['lgk'] |
| LG_TRIALS = argv['lgt'] |
| MAX_LG_N = argv['lgN'] # FOR SETTING NUMBER OF DISTINCTS |
| experiment = ErrorCardinalityProfile(SKETCH_LGK, LG_TRIALS, MAX_LG_N) |
| experiment.run() |
| |
| |
| |
| if __name__ == "__main__": |
| parser = argparse.ArgumentParser(description='Error-cardinality profile of HyperLogLog') |
| parser.add_argument('-lgk', '--lgk', help='Log2(k) value for the number of buckets in the sketch.',type=int, required=True) |
| parser.add_argument('-lgt', '--lgt', help='Log2(T) value for number of trials t.',type=int, required=True) |
| parser.add_argument('-lgN', '--lgN', help='Largest permissible log2(.) value for input size.',type=int, required=True) |
| args = vars(parser.parse_args()) |
| main(args) |
