def __init__(self, raw_data_options, metric_tags):
        """
        Args:
            raw_data_options: dj_pb.RawDataOptions. A protobuf containing
                optional stats options and arguments.
        """
        assert isinstance(raw_data_options, dj_pb.RawDataOptions)
        self._sample_reservoir = []
        self._sample_receive_num = 0
        self._reservoir_length = 10
        self._tags = copy.deepcopy(metric_tags)
        for k in list(self._tags.keys()):
            if k not in ('application_id', 'partition'):
                self._tags.pop(k)
        allowed_fields = {'label', 'type', 'joined'}
        optional_fields = set(raw_data_options.optional_fields)
        # prevent from adding too many fields to ES index
        self._stat_fields = optional_fields & allowed_fields
        self._sample_rate = Config.DATA_JOIN_METRICS_SAMPLE_RATE
        self._kind_map = {'unjoined': -1,
                          'fake': 0,
                          'joined': 1}
        # reversed map
        self._map_kind = {v: k for k, v in self._kind_map.items()}
        self._stats = {
            joined: defaultdict(int) for joined in self._kind_map.values()
        }

        Args:
            raw_data_options: dj_pb.RawDataOptions. A protobuf containing
                optional stats options and arguments.
        
fedlearner/data_join/joiner_impl/optional_stats.py