<jupyter_start><jupyter_text>IBM Streams Kafka sample applicationThis sample demonstrates how to create a Streams Python application that connects to a Kafka broker by using a consumer group and uses partitioned parallel processing of fetched messages.In this notebook, you'll see examples of how to: 1. [Setup your data connections](setup) 1. [Create the consumer application](create_1) 1. [Create a simple producer application](create_2) 1. [Submit the applications](launch) 1. [Connect to the running consumer application to view data](view) 1. [Stop the applications](cancel) Overview**About the sample**The main goal of the sample is to show how to connect to a Kafka broker and how to create a Kafka consumer group with downstream parallel processing of the fetched messages. A consumer group is mostly used to consume partitioned topics with multiple consumers sharing the partitions. The messages are typically distributed to the partitions by using a *key*. When keyed messages are processed in parallel after reception by a consumer group it is desired to stick a the message keys to one channel only. To achieve this, we route the tuples by a hash of the key to the parallel channel.Consuming a single-partition topic with more than one consumers has no advantage as failed consumers are restarted by Streams nearly as quickly as a failover of the single partition to another consumer would take.For completion of this sample there is also a data generator that publishes data to the topic.**How it works**The Python application created in this notebook is submitted to the IBM Streams service for execution. Once the application is running in the service, you can connect to it from the notebook to retrieve the results. Documentation- [Kafka consumer groups](https://kafka.apache.org/documentation/intro_consumers)- [Streams Python development guide](https://ibmstreams.github.io/streamsx.documentation/docs/latest/python/)- [Streams Python API](https://streamsxtopology.readthedocs.io/)- [streamsx.kafka Python package](https://streamsxkafka.readthedocs.io/) 1. Setup 1.1 Add credentials for the IBM Streams serviceWith the cell below selected, click the "Connect to instance" button in the toolbar to insert the credentials for the service.See an example.  1.2 Configure the connection to the Kafka brokerKafka consumers and producers are configured by properties. They are described in the Kafka documentation in separate sections for [producers configs](https://kafka.apache.org/21/documentation.htmlproducerconfigs) and [consumer configs](https://kafka.apache.org/21/documentation.htmlconsumerconfigs).The operators of the underlying SPL toolkit set defaults for some properties. You can review these operator provided defaults in the [toolkit documentation](https://ibmstreams.github.io/streamsx.kafka/doc/spldoc/html/) under **Operators**.The most important setting is the `bootstrap.servers` configuration, which is required for botch consumers and producers. This config has the form```host1:port1,host2:port2,....```Since these servers are just used for the initial connection to discover the full cluster membership (which may change dynamically), this list need not contain the full set of servers (you may want more than one, though, in case a server is down).Other settings, like SSL related settings depend on the broker configuration.In the Kafka broker, create a partitioned topic, for example with three partitions. Enter the bootstrap servers and the topic name into the below cell.<jupyter_code>bootstrap_servers =  # add the bootstrap servers here
topic =  # add the Kafka topic here, it should have multiple partitions

consumer_configs = dict()
consumer_configs['bootstrap.servers'] = bootstrap_servers
# add more consumer relevant configs if required
# Note: The group.id config is not required here. This is done on Python API level later<jupyter_output><empty_output><jupyter_text>1.3 Install or upgrade the `streamsx.kafka` packageRun the cell below to upgrade to the latest version of the `streamsx.kafka` package or to install the package.<jupyter_code>import sys
!{sys.executable} -m pip install --user --upgrade streamsx.kafka

# When you need to install a specific version of the package, run this line instead:
#!{sys.executable} -m pip install --user streamsx.kafka==somever
import streamsx.kafka as kafka
import streamsx.topology.context as context
print ("INFO: streamsx package version: " + context.__version__)
print ("INFO: streamsx.kafka package version: " + kafka.__version__)<jupyter_output><empty_output><jupyter_text>2. Create the consumer applicationThis application subscribes to a Kafka topic by using a consumer group (multiple consumers that share a group identifier).We assume that the messages we fetch from the topic, are JSON formatted with the content like```{"sensor_id": "sensor_4545", "value": 3567.87, "ts": 1559029421}``` All Streams applications start with a Topology object, so start by creating one:<jupyter_code>from streamsx.topology.topology import Topology
from streamsx.topology.context import submit, ContextTypes
from streamsx.topology.topology import Routing
from streamsx.topology.schema import StreamSchema
from streamsx.kafka.schema import Schema

consumer_topology = Topology (name = "ConsumerGroupSample")<jupyter_output><empty_output><jupyter_text>2.1 Create the consumer groupFrom the Kafka broker we fetch keyed messages, where the message type is a string. In addition to it we want to fetch the message meta data, like partition number, message timestamp, and other.That's why we specify `Schema.StringMessageMeta` as the schema for the created Stream.This schema is a structured schema that defines following attributes:- message(str) - the message content- key(str) - the key for partitioning- topic(str) - the Event Streams topic- partition(int) - the topic partition number (32 bit)- offset(int) - the offset of the message within the topic partition (64 bit)- messageTimestamp(int) - the message timestamp in milliseconds since epoch (64 bit)Create the stream `received` by subscribing to the Kafka topic, parallelize the source with `set_parallel` and combine the parallel streams with `end_parallel`. The result is a stream created by a consumer group with three consumers.<jupyter_code>consumerSchema = Schema.StringMessageMeta
received = kafka.subscribe(
    consumer_topology,
    topic=topic,
    schema=consumerSchema,
    group='my_consumer_group', # when not specified it is the job name, concatenated with the topic
    kafka_properties=consumer_configs,
    name="SensorSubscribe"
    ).set_parallel(3).end_parallel()<jupyter_output><empty_output><jupyter_text>2.2 Parallelize message processing with schema transformParallelize processing in four parallel channels, routing to the channels is hash based.We need a consistent hash function that calculates a hash from a string.    Warning: The Python hash function cannot be used as a consistent hash function as it adds random salt to the hash calculation. Its result is different after a process restarted. When a processing element in the Streams application is re-launched, the results of hash change, so that the routing to the parallel channel changes.<jupyter_code># calculate a hash code of a string in a consistent way
# needed for partitioned parallel streams
def string_hashcode(s):
    h = 0
    for c in s:
        h = (31 * h + ord(c)) & 0xFFFFFFFF
    return ((h + 0x80000000) & 0xFFFFFFFF) - 0x80000000


# start another parallel region partitioned by message key,
# so that each key always goes into the same parallel channel
receivedParallelPartitioned = received.parallel(
    4,
    routing=Routing.HASH_PARTITIONED,
    func=lambda x: string_hashcode(x['key']))<jupyter_output><empty_output><jupyter_text>Define a new schema by extending `typing.NamedTuple` and a function that parses the JSON of the messages and maps a couple of attributes of `Schema.StringMessageMeta` to the new schema.<jupyter_code>import typing
import json
class SensorMessage(typing.NamedTuple):
    sensor_id: str
    reading: float
    ts: int                # timestamp of the sensor measurement
    partition: int
    messageTimestamp: int  # timestamp of the Kafka message


# parses the JSON in the message and adds the attributes to a tuple
def parse_json_to_tuple(tuple):
    # the tuple is passed in as dict, the output tuple is typing.NamedTuple
    messageAsDict = json.loads(tuple['message']) # {"sensor_id": "sensor_4545", "value": 3567.87, "ts": 1559029421}
    # 5 required positional arguments: 'sensor_id', 'reading', 'ts', 'partition', and 'messageTimestamp'
    return SensorMessage(
        messageAsDict['sensor_id'],
        messageAsDict['value'],
        messageAsDict['ts'],
        tuple['partition'],
        tuple['messageTimestamp']
    )

# map the parallelized stream to the new schema
receivedParallelPartitionedParsed = receivedParallelPartitioned.map(
    func=parse_json_to_tuple,
    name='ParseMsgJson',
    schema=SensorMessage)

# validate by removing negative and zero values from the streams,
# pass only positive values and timestamps
receivedValidated = receivedParallelPartitionedParsed.filter(
    # remember, the tuple tup is passed as a Python named tuple, not as a dict
    lambda tup: (tup.reading > 0) and (tup.ts > 0),
    name='Validate')

# end parallel processing and (dummy) filter as it is not possible to create a view on a combined stream directly
parallelEnd = receivedValidated.end_parallel().filter(lambda x: x)<jupyter_output><empty_output><jupyter_text>2.3 Create a `View` to preview the tuples on the `Stream` A `View` is a connection to a `Stream` that becomes activated when the application is running. We examine the data from within the notebook in [section 5](view), below.<jupyter_code>streamView = parallelEnd.view(name="ValidatedSensorData",
                              description="Validated Sensor data")<jupyter_output><empty_output><jupyter_text>2.4 Define outputThe `parallelEnd` stream is our final result. We will use `Stream.publish()` to make this stream available to other Streams applications.If you want to send the stream to another database or system, you would use a sink function and invoke it using `Stream.for_each`. You can also use the functions of other Python packages to send the stream to other systems, for example the eventstore.<jupyter_code>import json
# publish results as JSON
parallelEnd.publish(topic="SensorData",
                    schema=json,
                    name="PublishSensors")

# Other options include:
# invoke another sink function:
# parallelEnd.for_each (func=send_to_db)
# parallelEnd.print()<jupyter_output><empty_output><jupyter_text>3. Create a simple producer applicationTo make the consumer application work, we need to publish some data to the topic. Therefore we create another, less complicated application that publishs data to the Kafka broker. 3.1 Define a data generator for the messages and a source stream<jupyter_code>import random
import time
import json
from datetime import datetime


# Define a callable source for data that we push into Event Streams
class SensorReadingsSource(object):
    def __call__(self):
        # This is just an example of using generated data,
        # Here you could connect to db
        # generate data
        # connect to data set
        # open file
        i = 0
        while(i < 500000):
            time.sleep(0.001)
            i = i + 1
            sensor_id = random.randint(1, 100)
            reading = {}
            reading["sensor_id"] = "sensor_" + str(sensor_id)
            reading["value"] = random.random() * 3000
            reading["ts"] = int(datetime.now().timestamp())
            yield reading

producer_topology = Topology("SensorProducer")

# create the data and map them to the attributes 'message' and 'key' of the
# 'Schema.StringMessage' schema for Kafka, so that we have messages with keys
sensorStream = producer_topology.source(
    SensorReadingsSource(),
    "RawDataSource"
    ).map(
        func=lambda reading: {'message': json.dumps(reading),
                              'key': reading['sensor_id']},
        name="ToKeyedMessage",
        schema=Schema.StringMessage)<jupyter_output><empty_output><jupyter_text>3.2 Publish the data to the Kafka topicFor advanced producer configurations please review the [producer configs section](https://kafka.apache.org/21/documentation.htmlproducerconfigs) of the Kafka documentation. Here we setup only the `bootstrap.servers` as used for the consumers. For a Kafka basic install out of the box this is sufficient. When you have configured authentication or other security options for the consumer, you must configure the same options also for the producer.<jupyter_code>producer_configs = dict()
producer_configs['bootstrap.servers'] = bootstrap_servers
# add more producer relevant configs if required

kafkaSink = kafka.publish(
    sensorStream,
    topic=topic,
    kafka_properties=producer_configs,
    name="SensorPublish")<jupyter_output><empty_output><jupyter_text>4. Submit both applications to the Streaming Analytics ServiceA running Streams application is called a *job*. By submitting the topologies we create two independent jobs.<jupyter_code># Disable SSL certificate verification if necessary
cfg[context.ConfigParams.SSL_VERIFY] = False

# submit consumer topology as a Streams job
consumer_submission_result = submit(ContextTypes.DISTRIBUTED, consumer_topology, cfg)
consumer_job = consumer_submission_result.job
if consumer_job:
    print("JobId of consumer job: ", consumer_job.id , "\nJob name: ", consumer_job.name)
# submit producer topology as a Streams job
producer_submission_result = submit(ContextTypes.DISTRIBUTED, producer_topology, cfg)
producer_job = producer_submission_result.job
if producer_job:
    print("JobId of producer job: ", producer_job.id , "\nJob name: ", producer_job.name)<jupyter_output><empty_output><jupyter_text>5. Use a `View` to access data from the jobNow that the job is started, use the `View` object you created in [step 2.3](create_view) to start retrieving data from a `Stream`.<jupyter_code># Connect to the view and display 20 samples of the data
queue = streamView.start_data_fetch()
try:
    for val in range(20):
        print(queue.get())    
finally:
    streamView.stop_data_fetch()<jupyter_output><empty_output><jupyter_text>5.1 Display the results in real timeCalling `View.display()` from the notebook displays the results of the view in a table that is updated in real-time.<jupyter_code># Display the results for 60 seconds
streamView.display(duration=60)<jupyter_output><empty_output><jupyter_text>4.2 See job status You can view job status and logs by going to **My Instances** > **Jobs**. Find your job based on the id printed above.Retrieve job logs using the "Download jobs" action from the job's context menu.To view other information about the job such as detailed metrics, access the Streams Console.  Go to **My Instances** > **Provisioned Instances**. Select the Streams instance and open the URL listed under *externalConsoleEndpoint* or *serviceConsoleEndpoint*.  6. Cancel the jobsThis cell generates widgets you can use to cancel the jobs.<jupyter_code># cancel the jobs in the IBM Streams service interactively
producer_submission_result.cancel_job_button()
consumer_submission_result.cancel_job_button()<jupyter_output><empty_output><jupyter_text>You can also interact with the job through the [Job](https://streamsxtopology.readthedocs.io/en/stable/streamsx.rest_primitives.htmlstreamsx.rest_primitives.Job) object returned from `producer_submission_result.job` and `consumer_submission_result.job`For example, use `producer_job.cancel()` to cancel the running producer job directly.<jupyter_code># cancel the jobs directly using the Job objects

# producer_job.cancel()
# consumer_job.cancel()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.07049591964846202, 'fraction_numerical': 0.009478970495919649, 'mean_word_length': 4.876429361859093, 'pattern_counts': {'":': 6, '<': 62, '<?xml version=': 0, '>': 65, 'https://': 9, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '15', 'dir': 'jupyter-structured-clean-dedup', 'id': '109421', 'license': 'Apache-2.0', 'n_tokens_mistral': 4264, 'n_tokens_neox': 4109, 'n_words': 1793, 'path': 'Streams-KafkaParallelSample.ipynb', 'repo_name': 'RolefH/sample.starter_notebooks', 'size': '15930'}