Only call the WriteAnalysisCacheToFS PTransform in Transform if the i…

…nput dictionary isn't empty.

PiperOrigin-RevId: 339255496

RELEASE.md

@@ -37,6 +37,8 @@
     tfx/examples.
 *   Fixed the run_component script.
 *   Stopped depending on `WTForms`.
+*   Fixed an issue with Transform cache and beam 2.24-2.25 in an interactive
+    notebook that caused it to fail.
 
 ### For pipeline authors
 

tfx/components/transform/executor.py

@@ -1125,12 +1125,15 @@ def _RunBeamImpl(self, analyze_data_list: List[_Dataset],
                     full_span_cache_dir,
                     os.path.join(output_cache_dir, span_cache_dir.key))
 
-          (cache_output
-           | 'WriteCache' >> analyzer_cache.WriteAnalysisCacheToFS(
-               pipeline=pipeline,
-               cache_base_dir=output_cache_dir,
-               sink=self._GetCacheSink(),
-               dataset_keys=full_analyze_dataset_keys_list))
+          # TODO(b/157479287, b/171165988): Remove this condition when beam 2.26
+          # is used.
+          if cache_output:
+            (cache_output
+             | 'WriteCache' >> analyzer_cache.WriteAnalysisCacheToFS(
+                 pipeline=pipeline,
+                 cache_base_dir=output_cache_dir,
+                 sink=self._GetCacheSink(),
+                 dataset_keys=full_analyze_dataset_keys_list))
 
         if compute_statistics or materialization_format is not None:
           # Do not compute pre-transform stats if the input format is raw proto,