code: def from_kafka_batched(
      topic,
      consumer_params,
      poll_interval="1s",
      npartitions=None,
      refresh_partitions=False,
      start=False,
      dask=False,
      max_batch_size=10000,
      keys=False,
      engine=None,
      **kwargs,
  ):
      """ Get messages and keys (optional) from Kafka in batches
  
      Uses the confluent-kafka library,
      https://docs.confluent.io/current/clients/confluent-kafka-python/
  
      This source will emit lists of messages for each partition of a single given
      topic per time interval, if there is new data. If using dask, one future
      will be produced per partition per time-step, if there is data.
  
      Checkpointing is achieved through the use of reference counting. A reference
      counter is emitted downstream for each batch of data. A callback is
      triggered when the reference count reaches zero and the offsets are
      committed back to Kafka. Upon the start of this function, the previously
      committed offsets will be fetched from Kafka and begin reading form there.
      This will guarantee at-least-once semantics.
  
      Parameters
      ----------
      topic: str
          Kafka topic to consume from
      consumer_params: dict
          | Settings to set up the stream, see
          | https://docs.confluent.io/current/clients/confluent-kafka-python/#configuration
          | https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
          | Examples:
          | bootstrap.servers: Connection string(s) (host:port) by which to reach Kafka
          | group.id: Identity of the consumer. If multiple sources share the same
          | group, each message will be passed to only one of them.
      poll_interval: number
          Seconds that elapse between polling Kafka for new messages
      npartitions: int (None)
          | Number of partitions in the topic.
          | If None, streamz will poll Kafka to get the number of partitions.
       refresh_partitions: bool (False)
          | Useful if the user expects to increase the number of topic partitions on the
          | fly, maybe to handle spikes in load. Streamz polls Kafka in every batch to
          | determine the current number of partitions. If partitions have been added,
          | streamz will automatically start reading data from the new partitions as well.
          | If set to False, streamz will not accommodate adding partitions on the fly.
          | It is recommended to restart the stream after decreasing the number of partitions.
      start: bool (False)
          Whether to start polling upon instantiation
      max_batch_size: int
          The maximum number of messages per partition to be consumed per batch
      keys: bool (False)
          | Whether to extract keys along with the messages.
          | If True, this will yield each message as a dict:
          | {'key':msg.key(), 'value':msg.value()}
      engine: str (None)
          | If engine is set to "cudf", streamz reads data (messages must be JSON)
          | from Kafka in an accelerated manner directly into cuDF (GPU) dataframes.
          | This is done using the RAPIDS custreamz library.
  
          | Please refer to RAPIDS cudf API here:
          | https://docs.rapids.ai/api/cudf/stable/
  
          | Folks interested in trying out custreamz would benefit from this
          | accelerated Kafka reader. If one does not want to use GPUs, they
          | can use streamz as is, with the default engine=None.
  
          | To use this option, one must install custreamz (use the
          | appropriate CUDA version recipe & Python version)
          | using a command like the one below, which will install all
          | GPU dependencies and streamz itself:
  
          | conda install -c rapidsai-nightly -c nvidia -c conda-forge \
          | -c defaults custreamz=0.15 python=3.7 cudatoolkit=10.2
  
          | More information at: https://rapids.ai/start.html
  
      Important Kafka Configurations
      By default, a stream will start reading from the latest offsets
      available. Please set 'auto.offset.reset': 'earliest' in the
      consumer configs, if the stream needs to start processing from
      the earliest offsets.
  
      Examples
      ----------
  
      >>> source = Stream.from_kafka_batched('mytopic',
      ...           {'bootstrap.servers': 'localhost:9092',
      ...            'group.id': 'streamz'})  # doctest: +SKIP
  
      """
      if dask:
          from distributed.client import default_client
  
          kwargs["loop"] = default_client().loop
      source = FromKafkaBatched(
          topic,
          consumer_params,
          poll_interval=poll_interval,
          npartitions=npartitions,
          refresh_partitions=refresh_partitions,
          max_batch_size=max_batch_size,
          keys=keys,
          engine=engine,
          **kwargs,
      )
      if dask:
          source = source.scatter()
  
      if start:
          source.start()
  
      if engine == "cudf":  # pragma: no cover
          return source.starmap(get_message_batch_cudf)
      else:
          return source.starmap(get_message_batch)
  
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 4 for calling a global function with example usage: from_kafka_batched(topic, consumer_params, poll_interval, npartitions, refresh_partitions, start, dask, max_batch_size, keys, engine) and returns: source, source