In the consumer, add filtering:
if event_type == "purchase":
# Only process purchase events
cursor.execute(insert_query, (event_id, event_type, user_id, event_timestamp, country))
conn.commit()
In the consumer, add filtering:
if event_type == "purchase":
# Only process purchase events
cursor.execute(insert_query, (event_id, event_type, user_id, event_timestamp, country))
conn.commit()