class_docstr: Base class for both local and cluster-based batched kafka processing
class_docstr_tok: ["Base","class","for","both","local","and","cluster-based","batched","kafka","processing"]
code: def __init__(
      self,
      topic,
      consumer_params,
      poll_interval="1s",
      npartitions=None,
      refresh_partitions=False,
      max_batch_size=10000,
      keys=False,
      engine=None,
      **kwargs,
  ):
      self.consumer_params = consumer_params
      # Override the auto-commit config to enforce custom streamz
      # checkpointing
      self.consumer_params["enable.auto.commit"] = "false"
      if "auto.offset.reset" not in self.consumer_params.keys():
          consumer_params["auto.offset.reset"] = "latest"
      self.topic = topic
      self.npartitions = npartitions
      self.refresh_partitions = refresh_partitions
      if self.npartitions is not None and self.npartitions <= 0:
          raise ValueError(
              "Number of Kafka topic partitions must be > 0."
          )
      self.poll_interval = convert_interval(poll_interval)
      self.max_batch_size = max_batch_size
      self.keys = keys
      self.engine = engine
      self.started = False
  
      super().__init__(**kwargs)
  
imports: import asyncio
  from glob import glob
  import queue
  import os
  import time
  from tornado import gen
  import weakref
  from .core import Stream, convert_interval, RefCounter, sync
label_desc: Use image node_id 1 to create a new FromKafkaBatched object from inherited base classes: Source with example: obj = FromKafkaBatched(topic, consumer_params, poll_interval, npartitions, refresh_partitions, max_batch_size, keys, engine)