02-kafka.service

[Unit]
Requires=zookeeper.service
After=zookeeper.service

[Service]
Type=simple
User=gb760
ExecStart=/bin/sh -c '/home/gb760/kafka/bin/kafka-server-start.sh /home/gb760/kafka/config/server.properties > /home/gb760/kafka/kafka.log 2>&1'
ExecStop=/home/gb760/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

03-zookeeper.service

[Unit]
Requires=network.target remote-fs.target
After=network.target remote-fs.target

[Service]
Type=simple
User=gb760
ExecStart=/home/gb760/kafka/bin/zookeeper-server-start.sh /home/gb760/kafka/config/zookeeper.properties
ExecStop=/home/gb760/kafka/bin/zookeeper-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

04-producer.py

from time import sleep
from json import dumps
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
                         value_serializer=lambda x:  
                         dumps(x).encode('utf-8'))

for e in range(1000):
    data = {'number' : e}
    producer.send('gb760', value=data)

05-consumer.py

from kafka import KafkaConsumer
from json import loads

consumer = KafkaConsumer(
    'gb760',
     bootstrap_servers=['localhost:9092'],
     auto_offset_reset='earliest',
     enable_auto_commit=True,
     group_id='my-group',
     value_deserializer=lambda x: x.decode('utf-8'))

for message in consumer:
    print(message)

06-commands.md

~/kafka/bin/kafka-topics.sh --create --topic gb760 --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

~/kafka/bin/kafka-console-producer.sh --topic gb760 --bootstrap-server localhost:9092

~/kafka/bin/kafka-console-consumer.sh --topic gb760 --bootstrap-server localhost:9092 --from-beginning