            [org.apache.beam.sdk Pipeline]
            [org.apache.beam.sdk.coders StringUtf8Coder CustomCoder Coder$Context KvCoder IterableCoder]
            [org.apache.beam.sdk.io
-            TextIO  TextIO$CompressionType]
+            TextIO  TextIO$CompressionType FileSystems FileBasedSink]
            [org.apache.beam.sdk.options PipelineOptionsFactory PipelineOptions]
            [org.apache.beam.runners.dataflow.options DataflowPipelineDebugOptions$DataflowClientFactory]
            [org.apache.beam.sdk.transforms

                       (Sessions/withGapDuration)
                       (Window/into))]
     (apply-transform pcoll transform window-schema options)))
  ([gap ^PCollection pcoll] (sliding-windows gap {} pcoll)))

  "Coerce from KV to Clojure MapEntry"
  [^KV kv]
  (let [v (.getValue kv)]
    (if (instance? Iterable v)
      (MapEntry. (.getKey kv) (seq v))
      (MapEntry. (.getKey kv) v))))

