class_docstr: Writes a cache object that can be read by ReadAnalysisCacheFromFS.
  
  Given a cache collection, this writes it to the configured directory.
  If the configured directory already contains cache, this will merge the new
  cache with the old.
  NOTE: This merging of cache is determined at beam graph construction time,
  so the cache must already exist there when constructing this.
class_docstr_tok: ["Writes","a","cache","object","that","can","be","read","by","ReadAnalysisCacheFromFS",".","Given","a","cache","collection",",","this","writes","it","to","the","configured","directory",".","If","the","configured","directory","already","contains","cache",",","this","will","merge","the","new","cache","with","the","old",".","NOTE",":","This","merging","of","cache","is","determined","at","beam","graph","construction","time",",","so","the","cache","must","already","exist","there","when","constructing","this","."]
docstr: Init method.
  
  Args:
    pipeline: A beam Pipeline.
    cache_base_dir: A str, the path that the cache should be stored in.
    dataset_keys: (Optional) An iterable of strings.
    sink: (Optional) A PTransform class that takes a path in its constructor,
      and is used to write the cache. If not provided this uses a GZipped
      TFRecord sink.
code: def __init__(
      self,
      pipeline: beam.Pipeline,
      cache_base_dir: str,
      dataset_keys: Optional[Iterable[DatasetKey]] = None,
      sink: Optional[object] = None,
  ):
      """Init method.
  
      Args:
        pipeline: A beam Pipeline.
        cache_base_dir: A str, the path that the cache should be stored in.
        dataset_keys: (Optional) An iterable of strings.
        sink: (Optional) A PTransform class that takes a path in its constructor,
          and is used to write the cache. If not provided this uses a GZipped
          TFRecord sink.
      """
      self.pipeline = pipeline
      self._cache_base_dir = cache_base_dir
      if dataset_keys is None:
          self._sorted_dataset_keys = None
      else:
          self._sorted_dataset_keys = sorted(dataset_keys)
      self._sink = sink
      if self._sink is None:
          # TODO(b/37788560): Possibly use Riegeli as a default file format once
          # possible.
          self._sink = _WriteToTFRecordGzip
  
imports: import os
  import pickle
  import re
  import sys
  from typing import Iterable, Mapping, List, Optional, Union, Tuple
  import apache_beam
  import tensorflow
  from tfx_bsl.types import tfx_namedtuple
label_desc: Use image node_id 1 to create a new WriteAnalysisCacheToFS object from inherited base classes: beam with example: obj = WriteAnalysisCacheToFS(pipeline, cache_base_dir, dataset_keys, sink)