title: Consume using KafkaIO in batch processing mode using Google Dataflow
body: <p>Google Dataflow job uses Apache Beam's <em>KafkaIO</em> library with <em>AvroIO</em> and Windowed Writes writing output to  ".avro" files in Google Cloud Storage bucket. However, it defaults to  <strong>Streaming</strong> as the processing job type on production data. </p>
  
  <p>Is it possible to consume data from Kafka topic using KafkaIO in Dataflow using <strong>Batch</strong> processing. This dataflow job does not require near real time processing (streaming). Is there a way to also insert the incoming records into BigQuery table without streaming inserts costs enabling the batch type processing.</p>
  
  <p>Batch processing with less frequent runs could work resulting in less memory, vCPUs and compute costs.</p>
  
  <p>As per:<a href="https://beam.apache.org/releases/javadoc/2.5.0/org/apache/beam/sdk/io/kafka/KafkaIO.html" rel="nofollow noreferrer">https://beam.apache.org/releases/javadoc/2.5.0/org/apache/beam/sdk/io/kafka/KafkaIO.html</a></p>
  
  <p><i>KafkaIO source returns unbounded collection of Kafka records as PCollection>. </i> </p>
  
  <p>Does it mean Kafka being unbounded source cannot be run in batch mode?</p>
  
  <p>Testing .withMaxNumRecords(1000000) condition runs the job in batch mode. However, to run the job in live incoming data, I need to remove this condition.</p>
  
  <p>I have tried using windowing and setting streaming mode options flag to false without success as in below code.</p>
  
  <pre class="lang-java prettyprint-override"><code>
  // not streaming mode
  options.setStreaming(false);
  
  ...
  PCollection&lt;String&gt; collection = p.apply(KafkaIO.&lt;String, String&gt;read()
                 .withBootstrapServers("IPADDRESS:9092")
                 .withTopic(topic)                                
                 .withKeyDeserializer(StringDeserializer.class)
                 .withValueDeserializer(StringDeserializer.class)                
                 .updateConsumerProperties(props)
                 .withConsumerFactoryFn(new ConsumerFactory())                            
  //             .withMaxNumRecords(1000000)
                 .withoutMetadata() 
         ).apply(Values.&lt;String&gt;create())
         .apply(Window.into(FixedWindows.of(Duration.standardDays(1))));
  
  
  ...
  //convert to Avro GenericRecord
  
  .apply("AvroToGCS", AvroIO.writeGenericRecords(AVRO_SCHEMA)
  .withWindowedWrites()
  .withNumShards(1)
  .to("gs://BUCKET/FOLDER/")
  .withSuffix(".avro"));
  
  </code></pre>
  
  <p>The code resulted in <strong>Streaming</strong> job type with 4 vCPUs and 1 worker for 9 minutes processing 1.8mn records. After this, I had to stop job (drain) to prevent costs.</p>
  
  <p>Enforcing the Batch processing in Dataflow on incoming data, is it possible to collect batch of records writing it as avro files and continue doing so until offset has caught up to <strong>latest</strong>.</p>
  
  <p>Any examples or sample code greatly appreciated.</p>