medium_ref_concurrent_queue_method.java

Long batchSize = 10L;

pipeline
  .apply("ReadItems", BigQueryIO.readTableRows().fromQuery(query))
  .apply("ParseTableRow", ParDo.of(new ParseTableRow()))
  .apply("ConcurrentQueueAggregate", ParDo.of(new DoFn<String, Iterable<String>>() {
      private Queue<String> queue = new ConcurrentLinkedQueue<>();

      @ProcessElement
      void processElement(ProcessContext c) {
          queue.add(c.element());
          if(queue.size() == batchSize) {
              c.output(queue);
              queue.clear();
          }
      }
  }))
  .apply("BatchPredict", ParDo.of(new BatchPredict()))
  .apply("StoreResults", TextIO.write());

pipeline.run();