Security and Privacy Program Governance and Strategy Questions

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
import redis, json, time

spark = SparkSession.builder.appName("consent-filter").getOrCreate()

# Read streaming source
raw = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers","kafka:9092") \
    .option("subscribe","events") \
    .load()

events = raw.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING) as json") \
    .selectExpr("key", "from_json(json, 'user_id STRING, payload MAP<STRING,STRING>, ts TIMESTAMP') as data") \
    .select("key", "data.*")

# Helper: refresh consent broadcast (runs on driver thread)
def load_consent():
    r = redis.Redis(host='redis', port=6379, db=0)
    # store consent as JSON map user_id -> boolean
    keys = r.hgetall("consent_map")
    return {k.decode(): v.decode()=="1" for k,v in keys.items()}

consent_map = spark.sparkContext.broadcast(load_consent())
LAST_REFRESH = time.time()

def enrich_and_filter(df, batch_id):
    global consent_map, LAST_REFRESH
    # refresh broadcast if older than 0.8s
    if time.time() - LAST_REFRESH > 0.8:
        consent_map.unpersist()
        consent_map = spark.sparkContext.broadcast(load_consent())
        LAST_REFRESH = time.time()

    # convert broadcast to DF for join (small)
    consent_df = spark.createDataFrame([(k, v) for k,v in consent_map.value.items()], ["user_id","consent"])
    joined = df.join(consent_df, df.user_id==consent_df.user_id, "left").filter(col("consent") == True)

    # write to Delta for storage (ACID, schema evolution) and to Kafka exactly-once using foreachBatch transactional producer
    joined.write.format("delta").mode("append").option("mergeSchema","true").save("/mnt/delta/events")

    # Write to Kafka using Spark's kafka sink to leverage exactly-once with checkpointing
    out = joined.selectExpr("CAST(user_id AS STRING) AS key", "to_json(struct(*)) AS value")
    out.write.format("kafka") \
       .option("kafka.bootstrap.servers","kafka:9092") \
       .option("topic","filtered-events") \
       .save()

query = events.writeStream.foreachBatch(enrich_and_filter) \
    .option("checkpointLocation","/mnt/checkpoints/consent-filter") \
    .start()
query.awaitTermination()