c3ca2456b96296326aa1d18f33ed72e8dd51ac5b,tensorflow_transform/analyzers.py,QuantilesCombiner,merge_accumulators,#QuantilesCombiner#Any#,1284

Before Change

  def merge_accumulators(self, summaries_iterable):
    graph_state = self._get_graph_state()
    result = None
    for summary in summaries_iterable:
      if summary is None:
        continue
      else:
        // This conversion is necessary to maintain compatibility with previous
        // versions, and will be removed in future (0.15).
        // TODO(b/127336397): Remove this.
        if isinstance(summary, (six.binary_type, six.text_type)):
          assert not summary
          continue
        with graph_state.lock:
          if self._use_core_quantile_ops:
            graph_state.thread_hostile_merge_inputs_callable(*summary)
            if result is not None:
              graph_state.thread_hostile_merge_inputs_callable(*result)  // pylint: disable=not-an-iterable
          else:
            graph_state.thread_hostile_merge_inputs_callable(summary)
            if result is not None:
              graph_state.thread_hostile_merge_inputs_callable(result)
          result = graph_state.thread_hostile_flush_summary_callable()
    return result

  def extract_output(self, summary):
    graph_state = self._get_graph_state()

After Change

    // there are less than N remaining. N=100.
    result = self.create_accumulator()
    // Make sure summaries is an iterator (so it remembers its position).
    summaries = iter(summaries)
    graph_state = self._get_graph_state()
    while True:
      batched_summaries = list(itertools.islice(summaries, 100))
      if not batched_summaries:
        break
      with graph_state.lock:
        if self._use_core_quantile_ops:
          graph_state.thread_hostile_merge_summary_callable(*result)
          for summary in batched_summaries:
            graph_state.thread_hostile_merge_summary_callable(*summary)
        else:
          graph_state.thread_hostile_merge_summary_callable(result)
          for summary in batched_summaries:
            graph_state.thread_hostile_merge_summary_callable(summary)
        result = graph_state.thread_hostile_flush_summary_callable()
    return result

  def extract_output(self, summary):

Instances