(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
Get messages and keys (optional) from Kafka in batches
Get messages and keys ( optional ) from Kafka in batches
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)
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 ) : 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 )