def METHOD_NAME(size):
  def _pipeline_runner():
    with beam.Pipeline(runner=DirectRunner()) as p:
      ts = TestStream().advance_watermark_to(0)
      all_elements = iter(range(size))
      watermark = 0
      while True:
        next_batch = list(itertools.islice(all_elements, 100))
        if not next_batch:
          break
        ts = ts.add_elements([(i, random.randint(0, 1000)) for i in next_batch])
        watermark = watermark + 100
        ts = ts.advance_watermark_to(watermark)
      ts = ts.advance_watermark_to_infinity()
      input_pc = p | ts | WindowInto(FixedWindows(100))
      for i in range(NUM_PARALLEL_STAGES):
        _build_serial_stages(input_pc, NUM_SERIAL_STAGES, i)
  return _pipeline_runner