parameters: (
          node,
          win,
          concurrency=None,
          pipeline=None)
return_statement: return process_node
docstring: group by window, for internal use
docstring_summary: group by window, for internal use
docstring_tokens: group by window for internal use
function: def node_window_by(
          node,
          win,
          concurrency=None,
          pipeline=None):
      """
      group by window, for internal use
      """
      plan = node.plan()
      scope = node.scope()
  
      shuffle = plan.shuffle(scope, [node])
  
      if concurrency is not None:
          shuffle.with_concurrency(concurrency)
      elif pipeline is not None and pipeline.estimate_concurrency:
          concurrency = node.size() / pipeline.size_per_concurrency
          shuffle.with_concurrency(concurrency)
  
      process_node = None
      if isinstance(win, window.DefaultWindow):
          shuffle_node = shuffle.node(0).distribute_as_batch()
          process_node = shuffle_node.set_debug_info("DefaultWindow")
      else:
          #FIXIT(zhangyuncong): only for test for now. Will fix before release.
          from bigflow.windowing import after_processing_time
          from bigflow.core import entity
  
          if win.trigger() is None:
              # just for test for now, at the final version it should be after watermark.
              shuffle_node = shuffle.window_by(win, after_processing_time.AfterProcessingTime(10)).node(0)
          else:
              shuffle_node = shuffle.window_by(win, win.trigger()).node(0)
  
          shuffle_node = shuffle_node.time_by(entity.PythonTimeReaderDelegator(lambda x: x[0]))
          process_node = shuffle_node.set_debug_info("Window")
  
      return process_node
function_tokens: def node_window_by ( node , win , concurrency = None , pipeline = None ) : plan = node . plan ( ) scope = node . scope ( ) shuffle = plan . shuffle ( scope , [ node ] ) if concurrency is not None : shuffle . with_concurrency ( concurrency ) elif pipeline is not None and pipeline . estimate_concurrency : concurrency = node . size ( ) / pipeline . size_per_concurrency shuffle . with_concurrency ( concurrency ) process_node = None if isinstance ( win , window . DefaultWindow ) : shuffle_node = shuffle . node ( 0 ) . distribute_as_batch ( ) process_node = shuffle_node . set_debug_info ( "DefaultWindow" ) else : #FIXIT(zhangyuncong): only for test for now. Will fix before release. from bigflow . windowing import after_processing_time from bigflow . core import entity if win . trigger ( ) is None : # just for test for now, at the final version it should be after watermark. shuffle_node = shuffle . window_by ( win , after_processing_time . AfterProcessingTime ( 10 ) ) . node ( 0 ) else : shuffle_node = shuffle . window_by ( win , win . trigger ( ) ) . node ( 0 ) shuffle_node = shuffle_node . time_by ( entity . PythonTimeReaderDelegator ( lambda x : x [ 0 ] ) ) process_node = shuffle_node . set_debug_info ( "Window" ) return process_node