<jupyter_start><jupyter_text>Login to https://community.pipeline.ai  Get Your User IdYou will need to fill in the unique values for the following:* ``  - 8 character id that uniquely identifies the PipelineAI user.  You will see the UserId in the upper right hand corner of the Settings tab after you login to [PipelineAI Community Edition](https://community.cloud.pipeline.ai)* `` - User-defined model/project name that uniquely identifies a model/project within your account.* `` - User-defined tag that uniquely identifies the model tag/version within a model/project![user-id](https://pipeline.ai/assets/img/user-id.png)<jupyter_code>user_id = YOUR_USER_ID
import numpy as np
import pylab
from tensorflow.contrib.kafka.python.ops import kafka_dataset_ops
from tensorflow.python.data.ops import iterator_ops
from tensorflow.python.framework import dtypes
from tensorflow.python.framework import errors
from tensorflow.python.ops import array_ops
from tensorflow.python.platform import test
import tensorflow as tf
from confluent_kafka import Producer
import json


num_samples = 100

x_train = np.random.rand(num_samples).astype(np.float32)
print(x_train)

noise = np.random.normal(scale=0.01, size=len(x_train))

y_train = x_train * 0.1 + 0.3 + noise
print(y_train)

pylab.plot(x_train, y_train, '.')

x_test = np.random.rand(len(x_train)).astype(np.float32)
print(x_test)

noise = np.random.normal(scale=.01, size=len(x_train))

y_test = x_test * 0.1 + 0.3 + noise
print(y_test)

pylab.plot(x_test, y_test, '.')
topic_name = user_id + '_linear'

p = Producer({'bootstrap.servers': 'kafka-cp-kafka-headless:9092'})

for x, y in zip(x_train, y_train):
    p.produce(topic_name, str((x, y)).encode('utf-8'))

if (p.flush() == 0):
    print('Success')
else:
    print('Failure')
def getSingleRecord(topic_name):
    topics = array_ops.placeholder(dtypes.string, shape=[None])
    num_epochs = array_ops.placeholder(dtypes.int64, shape=[])
    batch_size = array_ops.placeholder(dtypes.int64, shape=[])

    repeat_dataset = kafka_dataset_ops.KafkaDataset(
        servers="kafka-cp-kafka-headless:9092", 
        topics=[topic_name], 
        group="community", 
        eof=False).repeat(num_epochs)
    
    batch_dataset = repeat_dataset.batch(batch_size)

    iterator = iterator_ops.Iterator.from_structure(batch_dataset.output_types)
    init_op = iterator.make_initializer(repeat_dataset)
    get_next = iterator.get_next()
    
    with tf.Session() as sess:
        # Read from topic
        sess.run(
            init_op,
            feed_dict={
              topics: [topic_name],
              num_epochs: 1,
              batch_size: 5
            })
        try:
            print(sess.run(get_next).decode('utf-8'))
        except:
            print('No more messages to process.')

record = getSingleRecord(topic_name)
def getMultipleRecords(topic_name):
    topics = array_ops.placeholder(dtypes.string, shape=[None])
    num_epochs = array_ops.placeholder(dtypes.int64, shape=[])
    batch_size = array_ops.placeholder(dtypes.int64, shape=[])

    repeat_dataset = kafka_dataset_ops.KafkaDataset(
      servers="kafka-cp-kafka-headless:9092", 
      topics=[topic_name], 
      group="community", 
      eof=False).repeat(num_epochs)

    batch_dataset = repeat_dataset.batch(batch_size)

    iterator = iterator_ops.Iterator.from_structure(batch_dataset.output_types)
    init_op = iterator.make_initializer(repeat_dataset)
    get_next = iterator.get_next()

    with tf.Session() as sess:
        # Batched and repeated iteration through the stream.
        init_batch_op = iterator.make_initializer(batch_dataset)
        sess.run(
          init_batch_op,
            feed_dict={
            topics: [topic_name],
            num_epochs: 20,
            batch_size: 5
        })

        more_records = True
        while more_records:
           try:
               print(sess.run(get_next))
           except e:
               print('No more messages to process.')
               more_records = False

records = getMultipleRecords(topic_name)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10688140556368961, 'fraction_numerical': 0.011957052220595413, 'mean_word_length': 3.374599786552828, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 3, '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': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '573986', 'license': 'Apache-2.0', 'n_tokens_mistral': 1334, 'n_tokens_neox': 1297, 'n_words': 278, 'path': '07a_Datasets_Kafka.ipynb', 'repo_name': 'Radrangi/pipeainotebooks', 'size': '4098'}