kafka_stream.py

import uuid
from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'marcus',
    'start_date': datetime(2023, 9, 3, 10, 00)
}

def get_data():
    import requests
    url = "https://randomuser.me/api/"
    res = requests.get(url)
    res = res.json()
    res = res['results'][0]
    return res

def format_data(res):
    data = {}
    location = res['location']
    data['id'] = uuid.uuid4()
    data['first_name'] = res['name']['first']
    data['last_name'] = res['name']['last']
    data['gender'] = res['gender']
    data['address'] = f"{str(location['street']['number'])} {location['street']['name']}, " \
                      f"{location['city']}, {location['state']}, {location['country']}"
    data['post_code'] = location['postcode']
    data['email'] = res['email']
    data['username'] = res['login']['username']
    data['dob'] = res['dob']['date']
    data['registered_date'] = res['registered']['date']
    data['phone'] = res['phone']
    data['picture'] = res['picture']['medium']

    return data

def stream_data():
    import json
    from kafka import KafkaProducer
    import time
    import logging

    producer = KafkaProducer(bootstrap_servers=['localhost:9092'], max_block_ms=5000)
    curr_time = time.time()

    res = get_data()
    res = format_data(res)

    producer.send('users_created', json.dumps(res).encode('utf-8'))

    while True:
        if time.time() > curr_time + 60: # 1 min
            break
        try:
            res = get_data()
            res = format_data(res)

            producer.send('users_created', json.dumps(res))
        except Exception as e:
            logging.error(f'An error occured: {e}')
            continue

# stream_data()

with DAG('user_automation',
         default_args=default_args,
         schedule='@daily',
         catchup=False) as dag:

    streaming_task = PythonOperator(
        task_id='stream_data_from_api',
        python_callable=stream_data
    )