Extended Algorithms

class ExtendedHyperLogLogSketch: def __init__(self, b_m, b_s): self.b_m = b_m # Bucket bits self.b_s = b_s # Attribute bits self.m = 2 ** b_m self.s = 2 ** b_s # Core data structures self.registers = [0] * self.m # Max rank per bucket self.indicator_space = [0] * self.m # Collision resolution self.frequency_counts = [0] * self.m # Element count per bucket self.attribute_samples = [None] * self.m # Demographic info def _hash_function(self, data): # Double hashing for increased entropy inner_hash = hashlib.sha256(str(data).encode('utf-8')).hexdigest() outer_hash = hashlib.sha256(str(inner_hash).encode('utf-8')).hexdigest() hashed_ = '{:256b}'.format(int(outer_hash, 16)).replace(' ', '0') return hashed_ def update_sketch(self, event): user_id = event['id_to_count'] user_attr = event['attribute'] x = self._hash_function(user_id) # Extract components from hash j = int(str(x)[:self.b_m], 2) # Bucket index i = int(str(x)[self.b_m:self.b_m + self.b_s], 2) # Attribute indicator data = str(x)[self.b_m + self.b_s:] # Remaining bits # Calculate rank (trailing zeros + 1) w = 1 for c in reversed(data): if c == "0": w += 1 else: break # Update bucket with collision resolution if (self.registers[j] < w or (self.registers[j] == w and self.indicator_space[j] < i)): self.indicator_space[j] = i self.frequency_counts[j] = 1 self.attribute_samples[j] = user_attr elif self.registers[j] == w and self.indicator_space[j] == i: self.frequency_counts[j] += 1 # Handle attribute conflicts (last one wins, but could be weighted) if self.attribute_samples[j] != user_attr: self.attribute_samples[j] = user_attr self.registers[j] = max(self.registers[j], w)

class ExtendedHyperRealSketch: def __init__(self, b_m, b_s): self.b_m = b_m self.b_s = b_s self.m = 2 ** b_m self.s = 2 ** b_s # Core data structures (no indicator space needed for HR) self.registers = [1.0] * self.m # Min value per bucket self.frequency_counts = [0] * self.m # Element count per bucket self.attribute_samples = [None] * self.m # Demographic info def update_sketch(self, event): user_id = event['id_to_count'] user_attr = event['attribute'] x = self._hash_function(user_id) j = int(str(x)[:self.b_m], 2) # Bucket index # Normalize hash to [0,1] range int_val = int(hashlib.sha256(str(user_id).encode()).hexdigest()[:8], 16) w = int_val / (16 ** 8 - 1) # Update bucket (simpler than HLL - no collisions with float values) if self.registers[j] > w: self.frequency_counts[j] = 1 self.attribute_samples[j] = user_attr elif self.registers[j] == w: # Extremely rare with float values self.frequency_counts[j] += 1 if self.attribute_samples[j] != user_attr: self.attribute_samples[j] = user_attr self.registers[j] = min(self.registers[j], w)

def get_frequency_for_attr(self, attr=None): """Calculate cardinality estimates by demographic attribute""" # Collect attribute distribution from buckets attr_distribution = {} for count, attr_sample in zip(self.frequency_counts, self.attribute_samples): if attr_sample: if attr_sample in attr_distribution: attr_distribution[attr_sample] += count else: attr_distribution[attr_sample] = count # Normalize to probabilities total = sum(attr_distribution.values()) for attr_sample in attr_distribution: attr_distribution[attr_sample] /= total # Scale by total cardinality estimate total_cardinality = self.get_cardinality_estimate() for attr_sample in attr_distribution: attr_distribution[attr_sample] *= total_cardinality # Return specific attribute or full distribution if attr: return attr_distribution.get(attr, 0) else: return attr_distribution

# Simulation using NYC Parking Violations Dataset # Results for k=14 (16,384 buckets) results = {} for k in range(5, 20): ehll_sketch = ExtendedHyperLogLogSketch(b_m=k, b_s=k) ehr_sketch = ExtendedHyperRealSketch(b_m=k, b_s=k) # Process dataset with open('Parking_Violations_2019.csv', 'r') as f: for line in f: vehicle_id = extract_vehicle_id(line) registration_state = extract_state(line) ehll_sketch.update_sketch({ 'id_to_count': vehicle_id, 'attribute': registration_state }) ehr_sketch.update_sketch({ 'id_to_count': vehicle_id, 'attribute': registration_state }) # Calculate errors total_error_hll = calculate_total_error(ehll_sketch) total_error_hr = calculate_total_error(ehr_sketch) demographic_error_hll = calculate_demographic_error(ehll_sketch) demographic_error_hr = calculate_demographic_error(ehr_sketch) results[k] = { 'HLL_total_error': total_error_hll, 'HR_total_error': total_error_hr, 'HLL_demographic_error': demographic_error_hll, 'HR_demographic_error': demographic_error_hr }

def merge_extended_sketches(sketch1, sketch2): """Merge two extended sketches for cross-platform analytics""" merged_sketch = ExtendedHyperRealSketch(sketch1.b_m, sketch1.b_s) for i in range(sketch1.m): # For HyperReal: take minimum for union if sketch1.registers[i] <= sketch2.registers[i]: merged_sketch.registers[i] = sketch1.registers[i] merged_sketch.frequency_counts[i] = sketch1.frequency_counts[i] merged_sketch.attribute_samples[i] = sketch1.attribute_samples[i] else: merged_sketch.registers[i] = sketch2.registers[i] merged_sketch.frequency_counts[i] = sketch2.frequency_counts[i] merged_sketch.attribute_samples[i] = sketch2.attribute_samples[i] # Handle equal values (rare with floats) if sketch1.registers[i] == sketch2.registers[i]: merged_sketch.frequency_counts[i] = ( sketch1.frequency_counts[i] + sketch2.frequency_counts[i] ) # Attribute conflict resolution (could be more sophisticated) if sketch1.attribute_samples[i] != sketch2.attribute_samples[i]: # Random selection or weighted by frequency merged_sketch.attribute_samples[i] = random.choice([ sketch1.attribute_samples[i], sketch2.attribute_samples[i] ]) return merged_sketch