consumer.py

import json
from kafka import KafkaConsumer
from google.cloud import bigquery

if __name__ == "__main__":
    
    # bigwuery configuration
    bigquery_client = bigquery.Client()
    dataset_ref = bigquery_client.dataset('DATASET_NAME')
    table_ref = dataset_ref.table('TABLE_NAME')
    table = bigquery_client.get_table(table_ref)
    
    # set kafka topic name
    consumer = KafkaConsumer('KAFKA_TOPIC')
    
    # transformation and send data to bigquery
    for message in consumer:
        data = json.loads(message.value)
        user = data['payload']['after']
        stored_data = [
            (user['id'], user['firstname'], user['lastname'])
        ]
        bigquery_client.insert_rows(table, stored_data)