from . import SparkDriver
from kafka import KafkaProducer# type: ignore
import uuid

def publish(data):
    producer = KafkaProducer('localhost:2181')
    producer.send('test-topic' ,data)

class SparkProducer(SparkDriver):
    # Here we have all functions to be used for preprocessing and parsing/producing wanted data

    # At initalization we call super to get our hands in the spark driver
    # The SparkContext defined in SparkDriver will now be available with self.context
    def __init__(self) -> None:
        super()
    
    # After raw data is validated we are calling this to distribute data and apply computations
    def preprocess(self, data: list) -> None:
        # Spark not implemented yet
        #rdd = self.context.parallelize(list(range(0, 1000)))
        #rdd.map(publish)

        # Send data to kafka
        producer = KafkaProducer(bootstrap_servers=['localhost:9091'])
        for d in data:
            producer.send('test-topic', {"id": str(uuid.uuid4()), "data": d})
            producer.flush()
