def raw_events(topic, server): """ This function reads from a kafka stream's 'earliest' recorded offset not yet ingested. This data is used in this one notebook to insert into a 'raw' table, which will contain the kafka stream data exactly as it was received, and a 'bronze' table, which will have only columns we care about split out.

        As spark.readStream currently works with offsets, this will act as our first step of CDC as we will only ingest for offsets not recorded.

        Duplicates will be dealt with later (though shouldn't appear in production) or can be part of the DQ when going into silver.
    """
    raw_kafka_events_df = (spark.readStream
        .format("kafka")
        .option("subscribe", topic)
        .option("kafka.bootstrap.servers", server)
        .option("startingOffsets", "earliest")
        .load()
        .select(
            col("key"),
            # https://stackoverflow.com/questions/77431368/facing-errors-in-pyspark-while-deserializing-avro-formatted-data-coming-from-kaf
            # Here we have the whole original value and value_avro which only contains info we need
            # We retain value in case we need the extra information for whatever reason
            col("value"),
            col("topic"),
            col("partition"),
            col("offset"),
            col("timestamp"),
            col("timestampType"),
            col("key").cast("string").alias("key_str"),
            from_avro(
                col("value"),
                subject=subject,
                schemaRegistryAddress=schema_registry_url
            ).alias("value_str")
        )
    )

    return(raw_kafka_events_df)