ml_kafka_create_consumer_producer_3.py

for message in consumer:

    # Get the message body
    m=message.value

    # Turn the message back into a DataFrame
    df=DataFrame(m,index=[0])

    # Extract the Message from the DataFrame
    dff = pd.Series(df['MessageBody'])

    # Get a spam prediction for the message
    prediction = (model.predict(dff) > 0.5).astype(int)

    # Create a new message with the MessageID and the SpamFlag
    data = {'SpamFlag': prediction[0][0],
            'MessageID': m['MessageID']}
    
    # Print the message so you can see what is being sent
    print(f'Sending message: {data}')
    # Send the message containing the prediction to the predictions topic
    producer.send(topic="predictions",value=data)

    print(f'Sent record to topic at time {dt.datetime.utcnow()}')