<jupyter_start><jupyter_text>Write iris.csv to a Kafka topic<jupyter_code>from kafka import KafkaProducer
import avro
from avro import schema
from avro import io
import io, random
import csv

producer = KafkaProducer(bootstrap_servers=['kafka:9092'])
 
topic = "iris"

schema = avro.schema.Parse('{"type":"record","name":"Iris","namespace":"org.apache.samoa.avro.iris","fields":[{"name":"sepallength","type":"double"},{"name":"sepalwidth","type":"double"},{"name":"petallength","type":"double"},{"name":"petalwidth","type":"double"},{"name":"class","type":{"type":"enum","name":"Labels","symbols":["Iris-setosa","Iris-versicolor","Iris-virginica"]}}]}')

with open('/home/jovyan/work/datasets/iris-dataset/iris.data', 'r') as csvfile:
    reader = csv.reader(csvfile, delimiter=',', quotechar='|')
    writer = avro.io.DatumWriter(schema)
    for row in reader:
        bytes_writer = io.BytesIO()
        encoder = avro.io.BinaryEncoder(bytes_writer)
        writer.write({"sepallength":float(row[0]), "sepalwidth":float(row[1]), "petallength":float(row[2]), "petalwidth":float(row[3]), "class":row[4]}, encoder)
        raw_bytes = bytes_writer.getvalue()
        producer.send(topic, raw_bytes)<jupyter_output><empty_output><jupyter_text>Read iris.csv from a Kafka topic<jupyter_code>from kafka import KafkaConsumer
import avro
from avro import schema
from avro import io
import io
import csv

topic = 'iris'

consumer = KafkaConsumer(topic,
                         bootstrap_servers=['kafka:9092'],
                         auto_offset_reset='smallest',
                         consumer_timeout_ms=2000)

schema = avro.schema.Parse('{"type":"record","name":"Iris","namespace":"org.apache.samoa.avro.iris","fields":[{"name":"sepallength","type":"double"},{"name":"sepalwidth","type":"double"},{"name":"petallength","type":"double"},{"name":"petalwidth","type":"double"},{"name":"class","type":{"type":"enum","name":"Labels","symbols":["Iris-setosa","Iris-versicolor","Iris-virginica"]}}]}')

with open('iris.csv', 'w+') as csvfile:
    writer = csv.writer(csvfile, delimiter=',')
    writer.writerow(['sepallength','sepalwidth','petallength','petalwidth','class'])
    for msg in consumer:
        bytes_reader = io.BytesIO(msg.value)
        decoder = avro.io.BinaryDecoder(bytes_reader)
        reader = avro.io.DatumReader(schema)
        row = reader.read(decoder)
        writer.writerow([row['sepallength'], row['sepalwidth'], row['petallength'], row['petalwidth'], row['class']])<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.19873066243554144, 'fraction_numerical': 0.006743355811186037, 'mean_word_length': 5.853260869565218, 'pattern_counts': {'":': 39, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '255757', 'license': 'Apache-2.0', 'n_tokens_mistral': 848, 'n_tokens_neox': 816, 'n_words': 116, 'path': 'docker/docker-jupyter/examples/rw_iris_kafka.ipynb', 'repo_name': 'inovex/open-kubeyard', 'size': '2521'}