[WIP] Implement Accumulator Framework and Compound Accumulator for aggregation #47

preethiraghavan1 · 2021-06-18T18:43:58Z

Description

Implementing an abstract class for Accumulator and CompoundAccumulator class. Have not implemented serialization yet. Wanted to share this to see if this is in the right direction.

How has this been tested?

Unit tests

Checklist

[x ] I have followed the Contribution Guidelines and Code of Conduct
[ x] I have commented my code following the OpenMined Styleguide
I have labeled this PR with the relevant Type labels
[ x] My changes are covered by tests

…partition

…esent tuple 3. unpack arguments for readability

preethiraghavan1 · 2021-06-18T18:53:09Z

pipeline_dp/accumulator.py

+
+  @classmethod
+  def merge(cls, accumulators: typing.Iterable[
+    'Accumulator'] = None) -> 'Accumulator':


How do you like the idea of args type argument here.

merge(cls, *accumulators: 'Accumulator' = None)

A list need not be created when a single accumulator is involved.
But when a list is involved, then they have to unpack it merge(*list_of_acc)

May not be super helpful, but wanted to know your thoughts on this.

That's a nice idea. For now, as I foresee that the main usecase of calling of this function, there will be a list of accumulators. So let's keep for now iterable and if it's convenient in future we can change.

Got it! Thanks!

dvadym

Thanks it looks great! I left some comments

dvadym · 2021-06-19T17:09:18Z

pipeline_dp/accumulator.py

+
+  @classmethod
+  def merge(cls, accumulators: typing.Iterable[
+    'Accumulator'] = None) -> 'Accumulator':


That's a nice idea. For now, as I foresee that the main usecase of calling of this function, there will be a list of accumulators. So let's keep for now iterable and if it's convenient in future we can change.

dvadym · 2021-06-19T17:11:38Z

pipeline_dp/accumulator.py

+
+class Accumulator(abc.ABC):
+  """
+  Performs aggregations


Nit:

"""Base class for all accumulators.

Accumulators are objects that encapsulate aggregations and computations of differential private metrics.
"""

dvadym · 2021-06-19T17:12:24Z

pipeline_dp/accumulator.py

+    return cls(accumulators)
+
+  @abc.abstractmethod
+  def add_value(self, v):


Nit: v -> value

dvadym · 2021-06-19T17:13:36Z

pipeline_dp/accumulator.py

+  def add_value(self, v):
+    """
+    Adds the value to each of the accumulator constituting the
+    CompoundAccumulator


Please add '.' in the end

dvadym · 2021-06-19T17:13:49Z

pipeline_dp/accumulator.py

+    Adds the value to each of the accumulator constituting the
+    CompoundAccumulator
+    Args:
+      v: value to be added


Please add '.' in the end

dvadym · 2021-06-20T16:52:00Z

pipeline_dp/accumulator.py

+
+class CompoundAccumulator(Accumulator):
+  """
+  Performs aggregations for a compund accumulators


Please update the comment to

"""Accumulator for computing multiple metrics.

CompoundAccumulator contains one or more accumulators of other types for computing multiple metrics. For example it can contain [CountAccumulator, SumAccumulator]. CompoundAccumulator delegates all operation to the internal accumulators.
"""

dvadym · 2021-06-20T16:54:38Z

pipeline_dp/accumulator.py

+  Performs aggregations for a compund accumulators
+  """
+
+  def __init__(self, accumulators: typing.Iterable['Accumulator'] = None):


Nit: remove = None

dvadym · 2021-06-20T16:58:41Z

pipeline_dp/accumulator.py

+    self.accumulators = []
+    if accumulators:
+      # flatten the accumulators if the input is a CompundAccumulator
+      self.accumulators = [accumulator_expanded for accumulator in accumulators


An interesting idea to flatten accumulators (I didn't think about that). Let's do simpler for now and remove flattening.
I think that probably for now we will not have nested accumulators. When we need that, let's implement flattening

got it! Done!

dvadym · 2021-06-20T17:05:21Z

pipeline_dp/accumulator.py

+  def add_accumulator(self, accumulator: 'CompoundAccumulator') -> \
+    'CompoundAccumulator':
+    # merges the accumulators of the CompoundAccumulators
+    self.accumulators.extend(accumulator.accumulators)


Sry, I didn't explained that, the idea that we merge CompoundAcculators which contains accumulators of the same type and in the same order.
add_accumulator() will be called when we will have 1 accumulators, which have results from different worker machines.

E.g. let's assume that

compound_accumulator1.accumulators = [count_accumulator1, sum_accumulator1] compound_accumulator2.accumulators = [count_accumulator2, sum_accumulator2]

then compound_accumulator1.add_accumulator(compound_accumulator2) should do

count_accumulator1.add_accumulator(count_accumulator2) sum_accumulator1.add_accumulator(sum_accumulator2)

i.e. add_accumulator(acc) accumulates data from acc.

Thanks for explaining! I misunderstood the first time!

dvadym · 2021-06-20T17:16:34Z

tests/accumulator_test.py

+    self.assertEqual(computed_metrics, [4.5, 41])
+
+
+class MeanAccumulator(pipeline_dp.Accumulator):


I like idea with Mean and SumOfSquares accumulators for testing!

… and comment fixes

dvadym

Thanks! It looks great! I left some minor comments

dvadym · 2021-06-25T18:31:31Z

tests/accumulator_test.py

+    return self
+
+  def add_accumulator(self,
+                      accumulator: 'MeanAccumulator') -> 'MeanAccumulator':


Nit: MeanAccumulator -> SumOfSquaresAccumulator

pipeline_dp/accumulator.py

dvadym · 2021-06-25T18:37:13Z

pipeline_dp/accumulator.py

+
+  def add_accumulator(self, accumulator: 'CompoundAccumulator') -> \
+          'CompoundAccumulator':
+    # merges the accumulators of the CompoundAccumulators.


Nit:
""""Merges the accumulators of ...

The expectation ...
"""

dvadym · 2021-06-25T18:40:01Z

pipeline_dp/accumulator.py

+    # the expectation is that the input accumulators are of the same type and
+    # are in the same order.
+
+    if (len(accumulator.accumulators) != len(self.accumulators)


I like this thorough checks!

Nit: split these checks in 2:
1.The check of len()
2.The check of types:
for ... in zip():
if type() != type():
.... # and here we can tell which exactly types

This would allow us to provide more detailed error information

Let me know if the change looks good.

dvadym · 2021-06-25T18:44:09Z

pipeline_dp/accumulator.py

+    return self
+
+  def compute_metrics(self):
+    # Computes the metrics for individual accumulator


Nit:
'''Computes and returns a list of metrics computed by internal accumulators""""

pipeline_dp/accumulator.py

dvadym · 2021-06-25T18:56:27Z

pipeline_dp/accumulator.py

+  def add_accumulator(self, accumulator: 'Accumulator') -> 'Accumulator':
+    """
+      Merges the accumulator.
+      The difference between this and the merge function is that here it


After moving to merge functions to CompoundAccumulator. The comment "The difference... " is not needed. Please remove (maybe 1 line comment is enough)

Removed the line.

pipeline_dp/accumulator.py

dvadym · 2021-06-25T19:00:38Z

pipeline_dp/accumulator.py

+
+class CompoundAccumulator(Accumulator):
+  """
+    Accumulator for computing multiple metrics.


Nit:
"""Accumulator ...

dvadym · 2021-06-25T19:02:14Z

pipeline_dp/pipeline_operations.py

@@ -1,7 +1,6 @@
 """Adapters for working with pipeline frameworks."""

 import random
-import collections


Probably it's an incorrect merge in git. Please return back

dvadym · 2021-06-26T18:48:04Z

pipeline_dp/accumulator.py

+    differential private metrics.
+  """
+
+  @classmethod


During the reviewing PR I've realized that it's more convenient to have merge as a standalone function.

Could you please make merge a standalone function?

Made it a standalone function. What do you think of making it staticmethod in Accumulator?

dvadym

Thanks it looks great! Just some minor comments

dvadym · 2021-07-01T11:22:56Z

pipeline_dp/accumulator.py

+    CompoundAccumulator contains one or more accumulators of other types for
+    computing multiple metrics.
+    For example it can contain [CountAccumulator,  SumAccumulator].
+    CompoundAccumulator delegates all operation to the internal accumulators.


Nit: operations

dvadym · 2021-07-01T11:24:38Z

pipeline_dp/accumulator.py

+      "Accumulators should all be of the same type. Found accumulators of "
+      + f"different types: ({','.join(unique_accumulator_types)}).")
+
+  return reduce(lambda acc1, acc2: acc1.add_accumulator(acc2), accumulators)


Nice implementation!

dvadym · 2021-07-01T11:26:47Z

pipeline_dp/accumulator.py

+
+  Returns: Accumulator instance with merged values.
+  """
+  unique_accumulator_types = {type(accumulator).__name__ for accumulator in


Let's assume that add_accumulator checks types:

The correctness of add_accumulator depends on whether types the same and

add_accumulator might be called without merge

Nit: remove types checks

dvadym · 2021-07-01T11:28:53Z

pipeline_dp/accumulator.py

+  'Accumulator'] = None) -> 'Accumulator':
+  """Merges the accumulators.
+
+  Args:


Nit: the short description is enoughm remove starting from "Args:...

dvadym · 2021-07-01T11:29:01Z

pipeline_dp/accumulator.py

+
+
+def merge(accumulators: typing.Iterable[
+  'Accumulator'] = None) -> 'Accumulator':


Nit: remove = None

dvadym · 2021-07-01T11:30:37Z

pipeline_dp/accumulator.py

+
+  def add_accumulator(self, accumulator: 'CompoundAccumulator') -> \
+    'CompoundAccumulator':
+    """Merges the accumulators of the CompoundAccumulators.


Nit: add a blank line after this line

dvadym · 2021-07-01T11:35:07Z

pipeline_dp/accumulator.py

+        + f" Expected size = {len(self.accumulators)}"
+        + f" received size = {len(accumulator.accumulators)}.")
+
+    expected_type_order = ",".join([type(accumulator).__name__ for


Nit: Checking of types might be done in slighly shorter and more efficient way (no need to create lists):

for acc1, acc2 in zip(..., ...): if type(acc1) is not type(acc2): raise .... # here we can just provide the type of the 1st mismatched accumulators

Got it! Let me know if this looks good!

dvadym · 2021-07-01T11:57:18Z

tests/accumulator_test.py

+                     str(context.exception))
+
+  def test_adding_mismatched_accumulator_length_raises_exception(self):
+    mean_acc1 = MeanAccumulator().add_value(11)


Thanks, I like thorough testing!

Thanks for your thorough review!

dvadym · 2021-07-01T11:59:02Z

pipeline_dp/accumulator.py

+
+  @abc.abstractmethod
+  def add_accumulator(self, accumulator: 'Accumulator') -> 'Accumulator':
+    """Merges the accumulator to self and returns self.


Please add comment:

That sub-class implementation is responsible for checking that types of self and accumulator are the same.

dvadym · 2021-07-01T11:59:57Z

pipeline_dp/__init__.py

@@ -6,3 +6,4 @@
 from pipeline_dp.pipeline_operations import LocalPipelineOperations
 from pipeline_dp.pipeline_operations import BeamOperations
 from pipeline_dp.pipeline_operations import SparkRDDOperations
+from pipeline_dp.accumulator import CompoundAccumulator


Nit: remove from init.py

Explanation: CompoundAccumulator is not a part of Public API

Sorry missed that.

dvadym

Thanks for implementing Accumulator framework!

Preethi Raghavan and others added 30 commits May 19, 2021 18:01

Implementing Per partition - bounding of each contribution and cross …

53d5241

…partition

empty test case

161ee09

Comments

98efeaf

fixed the abstract method having beam operations in it.

8763009

format the files

d78506f

format the files

cf54652

comment fixes

2a3dc08

misc - Fixes for the comments on cl

a0051da

misc - Fixes for the comments on cl

7605b1c

added

6360c90

formatting the files

52d0577

I screwed up the formatting. Fixing now.

2ca2235

nit fix

7dfffb0

fixing merge conflict

54144fc

comment fixes

b5353bf

add sample_fixed_per_key_generator for lazy impl

33aed25

applying aggregate function after per partion bounding

992e0ef

adding types of the values

22b4077

resolving conflicts

0eb50cb

applying group by key and a few fixes

f59e2c7

function name change, comment change and extracting lambda to a function

ed0f632

test cases for sample_fixed_per_key and flat_map

591b620

1. create nested function, 2. rename variable pid to pid_pk_v to repr…

0546975

…esent tuple 3. unpack arguments for readability

comment in the nested function

b2f951b

removing the comment from the nested fn. It looks unnecessary.

411fc06

laziness test for sample_fixed_per_key and flat_map

e277898

Merge branch 'OpenMined:main' into main

d7567f8

Merge https://github.com/OpenMined/PipelineDP

963d943

Basic Accumulator and test cases

8d7f420

Adding comments

d41bef8

preethiraghavan1 requested a review from dvadym June 18, 2021 18:44

preethiraghavan1 commented Jun 18, 2021

View reviewed changes

Preethi Raghavan added 2 commits June 18, 2021 12:00

removing the unncessary formatting done

99ade2e

removing the unncessary formatting done

05717b4

dvadym reviewed Jun 20, 2021

View reviewed changes

Preethi Raghavan added 4 commits June 22, 2021 13:21

Compoud accumulator add_accumulator fix, remove flattening in merging…

041746c

… and comment fixes

Comment fix

12c08c9

Compound Accumulator cleanup

daefd5f

serialization

58abd55

preethiraghavan1 requested a review from dvadym June 25, 2021 17:45

dvadym reviewed Jun 25, 2021

View reviewed changes

dvadym reviewed Jun 26, 2021

View reviewed changes

Preethi Raghavan added 2 commits June 30, 2021 18:47

1. Merge as standalone function. 2. Comment fixes. 3. import fix

f029d51

format fix

3dbe745

preethiraghavan1 requested a review from dvadym July 1, 2021 01:59

dvadym reviewed Jul 1, 2021

View reviewed changes

Preethi Raghavan added 3 commits July 1, 2021 19:55

nit fixes

24d9869

merge function fixes

cb62ffd

merge function test fixes

d77c286

dvadym approved these changes Jul 2, 2021

View reviewed changes

dvadym merged commit 4d4fd45 into OpenMined:main Jul 2, 2021

		self.assertEqual(computed_metrics, [4.5, 41])


		class MeanAccumulator(pipeline_dp.Accumulator):



		def merge(accumulators: typing.Iterable[
		'Accumulator'] = None) -> 'Accumulator':

[WIP] Implement Accumulator Framework and Compound Accumulator for aggregation #47

[WIP] Implement Accumulator Framework and Compound Accumulator for aggregation #47

Conversation

preethiraghavan1 commented Jun 18, 2021

Description

How has this been tested?

Checklist

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dvadym left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dvadym left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dvadym left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dvadym left a comment

Choose a reason for hiding this comment