Contribution Bounding - per partition and cross partition bounding - continued from #26 pull request #32
Conversation
pipeline_dp/dp_engine.py
Outdated
| max_contributions_per_partition, | ||
| "Sample per (privacy_id, partition_key)") | ||
| # ((privacy_id, partition_key), [value]) | ||
| col = self._ops.map(col, lambda pid_pk: (pid_pk[0], aggregator_fn( |
There was a problem hiding this comment.
Please apply aggregator_fn per-partition contribution bounding, no need to do any aggregation after cross-partition bounding
Clarification:
eventually we need to aggregate per pk, we do it in 2 steps:
1.Aggregate per (pid, pk) (after per-partition contributions, this PR)
2.Aggregate per pk (after cross-partition contributions), but that's outside of the scope of this taskAggregation in 2 is slightly more complicated, because in 1 we can assume that data per key is in memory (because we've done bounding), but in 2 we can't that assume, so it's needed to use group by key.
That makes sense! Thanks!
Moved it to after per partition
pipeline_dp/dp_engine.py
Outdated
| @@ -49,3 +49,45 @@ def aggregate(self, col, params: AggregateParams, | |||
| # TODO: implement aggregate(). | |||
| # It returns input for now, just to ensure that the an example works. | |||
| return col | |||
|
|
|||
| def _bound_cross_partition_contributions(self, col, | |||
There was a problem hiding this comment.
Could you please rename to _bound_contributions (sry I realized that the name I provided is not correct)
pipeline_dp/dp_engine.py
Outdated
| max_contributions_per_partition, | ||
| "Sample per (privacy_id, partition_key)") | ||
| # ((privacy_id, partition_key), [value]) | ||
| col = self._ops.map(col, lambda pid_pk: (pid_pk[0], aggregator_fn( |
pipeline_dp/dp_engine.py
Outdated
| max_contributions_per_partition, | ||
| "Sample per (privacy_id, partition_key)") | ||
| # ((privacy_id, partition_key), [value]) | ||
| col = self._ops.map(col, lambda pid_pk: (pid_pk[0], aggregator_fn( |
There was a problem hiding this comment.
Nit (optional/style improvements): using map_values() instead of map() is simpler
There was a problem hiding this comment.
Agreed! Done
pipeline_dp/dp_engine.py
Outdated
| # Cross partition bounding | ||
| col = self._ops.map_tuple(col, lambda pid_pk, v: (pid_pk[0], | ||
| (pid_pk[1], v)), | ||
| "To (privacy_id, (partition_key, aggregator))") |
There was a problem hiding this comment.
Please update a stage name to "Rekey to ...".
Clarification:the operation of changing keys is needed very often and usually we use Rekey name for it.
| @@ -176,8 +188,19 @@ def keys(self, col, stage_name: str): | |||
| def values(self, col, stage_name: typing.Optional[str] = None): | |||
| return (v for k, v in col) | |||
|
|
|||
| def sample_fixed_per_key(self, col, n: int, stage_name: str): | |||
| pass | |||
| def sample_fixed_per_key(self, col, n: int, | |||
There was a problem hiding this comment.
Please add a test for this function
pipeline_dp/dp_engine.py
Outdated
| col = self._ops.sample_fixed_per_key(col, max_partitions_contributed, | ||
| "Sample per privacy_id") | ||
| # (privacy_id, [(partition_key, aggregator)]) | ||
| return self._ops.flat_map(col, lambda pid: [((pid[0], pk_v[0]), pk_v[1]) |
There was a problem hiding this comment.
Nit: lambda in this line is pretty complicated, maybe try a local function instead of lambda.
and please use lazy evaluations (there are 2 options "yield" or "(...)" generator)
pipeline_dp/dp_engine.py
Outdated
| max_contributions_per_partition: int, | ||
| aggregator_fn): | ||
| """ | ||
| Bounds the contribution by privacy_id in and cross partitions |
There was a problem hiding this comment.
Please add "." after this phrase and then after each args description and return description
pipeline_dp/dp_engine.py
Outdated
| self._unnest_cross_partition_bound_sampled_per_key, | ||
| "Unnest") | ||
|
|
||
| def _unnest_cross_partition_bound_sampled_per_key(self, pid): |
There was a problem hiding this comment.
I'd suggest to make it internal function of _bound_contributions, and no need to make comments then.
There was a problem hiding this comment.
I agree it makes it better to be a nested function. Style guide suggested that we use nested function only for closing over a variable, so, added as a module level. Moved it to inner function.
pipeline_dp/dp_engine.py
Outdated
| self._unnest_cross_partition_bound_sampled_per_key, | ||
| "Unnest") | ||
|
|
||
| def _unnest_cross_partition_bound_sampled_per_key(self, pid): |
There was a problem hiding this comment.
please rename pid to pid_pk_v (since it's not pid, but a tuple)
pipeline_dp/dp_engine.py
Outdated
|
|
||
| Returns: tuple of the form ((privacy_id, partition_key), values) | ||
|
|
||
| """ |
There was a problem hiding this comment.
please unpack arguments for readability:
pid, pk_values = pid_pk_v
| @@ -176,8 +188,19 @@ def keys(self, col, stage_name: str): | |||
| def values(self, col, stage_name: typing.Optional[str] = None): | |||
| return (v for k, v in col) | |||
|
|
|||
| def sample_fixed_per_key(self, col, n: int, stage_name: str): | |||
| pass | |||
| def sample_fixed_per_key(self, col, n: int, | |||
| @@ -133,6 +153,52 @@ def assert_laziness(operator, *args): | |||
| assert_laziness(self.ops.values) | |||
| assert_laziness(self.ops.count_per_element) | |||
|
|
|||
| def test_local_sample_fixed_per_key_requires_no_discarding(self): | |||
There was a problem hiding this comment.
please add
assert_laziness(self.ops.sample_fixed_per_key)
assert_laziness(self.ops.flat_map)
in test_laziness a few lines above
| @@ -133,6 +153,52 @@ def assert_laziness(operator, *args): | |||
| assert_laziness(self.ops.values) | |||
| assert_laziness(self.ops.count_per_element) | |||
|
|
|||
| def test_local_sample_fixed_per_key_requires_no_discarding(self): | |||
Description
Implementing Per partition - bounding of each contribution and cross partition. Added flat map in pipeline_operations.py to accommodate this functionality. Continuation of #26
How has this been tested?
Checklist