0d078991bf6a056be5579cda14dd1f305078eb7d,tensorflow_transform/beam/cached_impl_test.py,CachedImplTest,test_single_phase_run_twice,#CachedImplTest#,419

Before Change

          self.assertIn(key, cache_output)
          self.assertEqual(7, len(cache_output[key]))

    with beam_impl.Context(temp_dir=self.get_temp_dir()):
      with beam.Pipeline() as p:

        flat_data = p | "CreateInputData" >> beam.Create(
            list(itertools.chain(*input_data_dict.values())))

        // wrap each value in input_data_dict as a pcoll.
        input_data_pcoll_dict = {}
        for a, b in six.iteritems(input_data_dict):
          input_data_pcoll_dict[a] = p | a >> beam.Create(b)

        input_cache = p | analyzer_cache.ReadAnalysisCacheFromFS(
            self._cache_dir, list(input_data_dict.keys()))

        transform_fn_2, second_output_cache = (
            (flat_data, input_data_pcoll_dict, input_cache, input_metadata)
            | "AnalyzeAgain" >>
            (beam_impl.AnalyzeDatasetWithCache(preprocessing_fn)))

        dot_string = nodes.get_dot_graph(
            [analysis_graph_builder._ANALYSIS_GRAPH]).to_string()
        self.WriteRenderedDotFile(dot_string)

        transformed_dataset = (
            ((input_data_dict[span_1_key], input_metadata), transform_fn_2)
            | "TransformAgain" >> beam_impl.TransformDataset())
    transformed_data, unused_transformed_metadata = transformed_dataset
    beam_test_util.assert_that(
        transformed_data,
        beam_test_util.equal_to(expected_transformed_data),

After Change

    // 4 from analyzing 2 spans, and 2 from transform.
    self.assertEqual(_get_counter_value(p.metrics, "num_instances"), 6)
    self.assertEqual(_get_counter_value(p.metrics, "cache_entries_decoded"), 0)
    self.assertEqual(_get_counter_value(p.metrics, "cache_entries_encoded"), 14)
    self.assertEqual(_get_counter_value(p.metrics, "saved_models_created"), 2)

    with _TestPipeline() as p:

Instances