class_docstr: A Discretized Stream (DStream), the basic abstraction in Spark Streaming,
  is a continuous sequence of RDDs (of the same type) representing a
  continuous stream of data (see :class:`RDD` in the Spark core documentation
  for more details on RDDs).
  
  DStreams can either be created from live data (such as, data from TCP
  sockets, etc.) using a :class:`StreamingContext` or it can be
  generated by transforming existing DStreams using operations such as
  `map`, `window` and `reduceByKeyAndWindow`. While a Spark Streaming
  program is running, each DStream periodically generates a RDD, either
  from live data or by transforming the RDD generated by a parent DStream.
  
  DStreams internally is characterized by a few basic properties:
   - A list of other DStreams that the DStream depends on
   - A time interval at which the DStream generates an RDD
   - A function that is used to generate an RDD after each time interval
class_docstr_tok: ["A","Discretized","Stream","(","DStream",")",",","the","basic","abstraction","in","Spark","Streaming",",","is","a","continuous","sequence","of","RDDs","(","of","the","same","type",")","representing","a","continuous","stream","of","data","(","see",":","class",":","`","RDD","`","in","the","Spark","core","documentation","for","more","details","on","RDDs",")",".","DStreams","can","either","be","created","from","live","data","(","such","as",",","data","from","TCP","sockets",",","etc",".",")","using","a",":","class",":","`","StreamingContext","`","or","it","can","be","generated","by","transforming","existing","DStreams","using","operations","such","as","`","map","`",",","`","window","`","and","`","reduceByKeyAndWindow","`",".","While","a","Spark","Streaming","program","is","running",",","each","DStream","periodically","generates","a","RDD",",","either","from","live","data","or","by","transforming","the","RDD","generated","by","a","parent","DStream",".","DStreams","internally","is","characterized","by","a","few","basic","properties",":","-","A","list","of","other","DStreams","that","the","DStream","depends","on","-","A","time","interval","at","which","the","DStream","generates","an","RDD","-","A","function","that","is","used","to","generate","an","RDD","after","each","time","interval"]
docstr: Return a new DStream by applying incremental `reduceByKey` over a sliding window.
  
  The reduced value of over a new window is calculated using the old window's reduce value :
   1. reduce the new values that entered the window (e.g., adding new counts)
   2. "inverse reduce" the old values that left the window (e.g., subtracting old counts)
  
  `invFunc` can be None, then it will reduce all the RDDs in window, could be slower
  than having `invFunc`.
  
  Parameters
  ----------
  func : function
      associative and commutative reduce function
  invFunc : function
      inverse function of `reduceFunc`
  windowDuration : int
      width of the window; must be a multiple of this DStream's
      batching interval
  slideDuration : int, optional
      sliding interval of the window (i.e., the interval after which
      the new DStream will generate RDDs); must be a multiple of this
      DStream's batching interval
  numPartitions : int, optional
      number of partitions of each RDD in the new DStream.
  filterFunc : function, optional
      function to filter expired key-value pairs;
      only pairs that satisfy the function are retained
      set this to null if you do not want to filter
docstr_tok: ["Return","a","new","DStream","by","applying","incremental","`","reduceByKey","`","over","a","sliding","window",".","The","reduced","value","of","over","a","new","window","is","calculated","using","the","old","window","'s","reduce","value",":","1.","reduce","the","new","values","that","entered","the","window","(","e.g.",",","adding","new","counts",")","2",".","``","inverse","reduce","''","the","old","values","that","left","the","window","(","e.g.",",","subtracting","old","counts",")","`","invFunc","`","can","be","None",",","then","it","will","reduce","all","the","RDDs","in","window",",","could","be","slower","than","having","`","invFunc","`",".","Parameters","--","--","--","--","--","func",":","function","associative","and","commutative","reduce","function","invFunc",":","function","inverse","function","of","`","reduceFunc","`","windowDuration",":","int","width","of","the","window",";","must","be","a","multiple","of","this","DStream's","batching","interval","slideDuration",":","int",",","optional","sliding","interval","of","the","window","(","i.e.",",","the","interval","after","which","the","new","DStream","will","generate","RDDs",")",";","must","be","a","multiple","of","this","DStream","'s","batching","interval","numPartitions",":","int",",","optional","number","of","partitions","of","each","RDD","in","the","new","DStream",".","filterFunc",":","function",",","optional","function","to","filter","expired","key-value","pairs",";","only","pairs","that","satisfy","the","function","are","retained","set","this","to","null","if","you","do","not","want","to","filter"]
code: def reduceByKeyAndWindow(
      self: "DStream[Tuple[K, V]]",
      func: Callable[[V, V], V],
      invFunc: Optional[Callable[[V, V], V]],
      windowDuration: int,
      slideDuration: Optional[int] = None,
      numPartitions: Optional[int] = None,
      filterFunc: Optional[Callable[[Tuple[K, V]], bool]] = None,
  ) -> "DStream[Tuple[K, V]]":
      """
      Return a new DStream by applying incremental `reduceByKey` over a sliding window.
  
      The reduced value of over a new window is calculated using the old window's reduce value :
       1. reduce the new values that entered the window (e.g., adding new counts)
       2. "inverse reduce" the old values that left the window (e.g., subtracting old counts)
  
      `invFunc` can be None, then it will reduce all the RDDs in window, could be slower
      than having `invFunc`.
  
      Parameters
      ----------
      func : function
          associative and commutative reduce function
      invFunc : function
          inverse function of `reduceFunc`
      windowDuration : int
          width of the window; must be a multiple of this DStream's
          batching interval
      slideDuration : int, optional
          sliding interval of the window (i.e., the interval after which
          the new DStream will generate RDDs); must be a multiple of this
          DStream's batching interval
      numPartitions : int, optional
          number of partitions of each RDD in the new DStream.
      filterFunc : function, optional
          function to filter expired key-value pairs;
          only pairs that satisfy the function are retained
          set this to null if you do not want to filter
      """
      self._validate_window_param(windowDuration, slideDuration)
      if numPartitions is None:
          numPartitions = self._sc.defaultParallelism
  
      reduced = self.reduceByKey(func, numPartitions)
  
      if invFunc:
  
          def reduceFunc(t: datetime, a: Any, b: Any) -> Any:
              b = b.reduceByKey(func, numPartitions)
              r = (
                  a.union(b).reduceByKey(func, numPartitions)
                  if a
                  else b
              )
              if filterFunc:
                  r = r.filter(filterFunc)
              return r
  
          def invReduceFunc(t: datetime, a: Any, b: Any) -> Any:
              b = b.reduceByKey(func, numPartitions)
              joined = a.leftOuterJoin(b, numPartitions)
              return joined.mapValues(
                  lambda kv: invFunc(kv[0], kv[1])  # type: ignore[misc]
                  if kv[1] is not None
                  else kv[0]
              )
  
          jreduceFunc = TransformFunction(
              self._sc, reduceFunc, reduced._jrdd_deserializer
          )
          jinvReduceFunc = TransformFunction(
              self._sc, invReduceFunc, reduced._jrdd_deserializer
          )
          if slideDuration is None:
              slideDuration = self._slideDuration
          assert self._sc._jvm is not None
          dstream = self._sc._jvm.PythonReducedWindowedDStream(
              reduced._jdstream.dstream(),
              jreduceFunc,
              jinvReduceFunc,
              self._ssc._jduration(windowDuration),
              self._ssc._jduration(slideDuration),  # type: ignore[arg-type]
          )
          return DStream(
              dstream.asJavaDStream(), self._ssc, self._sc.serializer
          )
      else:
          return reduced.window(
              windowDuration, slideDuration
          ).reduceByKey(
              func, numPartitions  # type: ignore[arg-type]
          )
  
imports: import operator
  import time
  from itertools import chain
  from datetime import datetime
  from typing import Any, Callable, Generic, Hashable, Iterable, List, Optional, Tuple, TypeVar, Union, TYPE_CHECKING, cast, overload
  from py4j.protocol import Py4JJavaError
  from py4j.java_gateway import JavaObject
  from pyspark.storagelevel import StorageLevel
  from pyspark.streaming.util import rddToFileName, TransformFunction
  from pyspark.rdd import portable_hash, RDD
  from pyspark.resultiterable import ResultIterable
label_desc: Use image node_id 48 for calling the DStream obj's underlying member method code with example usage: obj.reduceByKeyAndWindow(func, invFunc, windowDuration, slideDuration, numPartitions, filterFunc) and returns: DStream, reduced, r, joined