tff.aggregators.UnweightedReservoirSamplingFactory

An UnweightedAggregationFactory for reservoir sampling values.

Inherits From: UnweightedAggregationFactory

tff.aggregators.UnweightedReservoirSamplingFactory(
    sample_size: int
)

The created tff.templates.AggregationProcess samples values placed at CLIENTS, and outputs the sample placed at SERVER.

The process has empty state. The measurements of this factory counts the number of non-finite (NaN or Inf values) leaves in the client values before sampling. Specifically, the returned measurements has the same structure as the client value, and every leaf node is a tf.int64 scalar tensor counting the number of clients having non-finite value in that leaf.

For example, suppose we are aggregating from three clients:

  client_value_1 = collections.OrderedDict(a=[1.0, 2.0], b=[1.0, np.nan])
  client_value_2 = collections.OrderedDict(a=[np.nan, np.inf], b=[1.0, 2.0])
  client_value_3 = collections.OrderedDict(a=[1.0, 2.0], b=[np.inf, np.nan])

Then measurements will be:

  collections.OrderedDict(a=tf.constant(1, dtype=int64),
                          b=tf.constant(2, dtype=int64)

For more about reservoir sampling see https://en.wikipedia.org/wiki/Reservoir_sampling

Methods

`create`

View source

create(
    value_type: factory.ValueType
) -> tff.templates.AggregationProcess

Creates a tff.aggregators.AggregationProcess without weights.

The provided value_type is a non-federated tff.Type object, that is, value_type.is_federated() should return False.

The returned tff.aggregators.AggregationProcess will be created for aggregation of values matching value_type placed at tff.CLIENTS. That is, its next method will expect type <S@SERVER, {value_type}@CLIENTS>, where S is the unplaced return type of its initialize method.

Args
`value_type`	A non-federated `tff.Type` (`value_type.is_federated()` returns `False`).

Returns
A `tff.templates.AggregationProcess`.