import requests

url = "https://randomuser.me/api/?inc=gender,name,nat,phone&format=prettyJSON"

response = requests.get(url)

if response.status_code == 200:
    data = response.json()
    print(data)  # Print the entire response
else:
    print(f"Error: {response.status_code}")

{'results': [{'gender': 'female', 'name': {'title': 'Madame', 'first': 'Viviane', 'last': 'Vidal'}, 'phone': '078 338 23 37', 'nat': 'CH'}], 'info': {'seed': 'a36cbb9b0b403a9a', 'results': 1, 'page': 1, 'version': '1.4'}}

data

{'results': [{'gender': 'female',
   'name': {'title': 'Madame', 'first': 'Viviane', 'last': 'Vidal'},
   'phone': '078 338 23 37',
   'nat': 'CH'}],
 'info': {'seed': 'a36cbb9b0b403a9a',
  'results': 1,
  'page': 1,
  'version': '1.4'}}

result = data['results'][0]  # Get the first user
row_data = {
        "Title": result['name']['title'],
        "First Name": result['name']['first'],
        "Nationality": result['nat'],
        "Last Name": result['name']['last'],
        "Gender": result['gender'],
        "Phone": result['phone']
    }

row_data

{'Title': 'Madame',
 'First Name': 'Viviane',
 'Nationality': 'CH',
 'Last Name': 'Vidal',
 'Gender': 'female',
 'Phone': '078 338 23 37'}

message = json.dumps(row_data).encode('utf-8')
message

b'{"Title": "Madame", "First Name": "Viviane", "Nationality": "CH", "Last Name": "Vidal", "Gender": "female", "Phone": "078 338 23 37"}'

from confluent_kafka import Producer

KAFKA_CONFIG = {"bootstrap.servers" : "localhost:9092",
        "client.id" : "random-user-producer"}

producer = Producer(KAFKA_CONFIG)

TOPIC = 'random-user-topic'
INTERVAL = 30

producer.produce(topic = TOPIC, value = message)
producer.flush()

0

import pandas as pd
import json
from confluent_kafka import Consumer, KafkaException, KafkaError

pd.options.display.float_format = '{:,.3f}'.format

# Konfigurasi Kafka Consumer
consumer_config = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "random-user-group",
    "auto.offset.reset": "earliest"
}

consumer = Consumer(consumer_config)

# Subscribe ke topik Kafka
topic_name = 'random-user-topic'

consumer.subscribe([topic_name])

consumer.subscribe(["topic1", "topic2"])

msg = consumer.poll(timeout=1.0)

msg

<cimpl.Message at 0x11eb41cc0>

msg.value().decode('utf-8')

'{"Title": "Mrs", "First Name": "Rosalinde", "Last Name": "Greb", "Gender": "female", "Phone": "0948-7585689"}'

# List awal untuk menyimpan data
random_user_data=[]

# Parsing data
random_user_data_point = json.loads(msg.value().decode('utf-8'))

random_user_data.append(random_user_data_point)

# Mengubah list menjadi DataFrame untuk analisis lebih lanjut
random_user_data_df = pd.DataFrame(random_user_data, columns=['Title', 'First Name', 'Last Name', 'Gender', 'Phone'])
random_user_data_df

# Count occurrences of each gender
gender_counts = random_user_data_df['Gender'].value_counts()

display(gender_counts)

Gender
male      6
female    3
Name: count, dtype: int64

try:
    # Diisi dengan kode yang mungkin menyebabkan error
    pass  # Misalnya, membuka file atau melakukan koneksi database
    
except ExceptionType:
    # Diisi dengan kode untuk menangani error
    pass  # Menangani error tertentu (misalnya, FileNotFoundError)
    
finally:
    # Diisi dengan kode yang ingin dijalankan terlepas dari apakah ada error atau tidak
    pass  # Menutup koneksi atau membersihkan resource

except KeyboardInterrupt:
    print("Stopped by user")

import pandas as pd
import json
from confluent_kafka import Consumer, KafkaError
import time
import matplotlib.pyplot as plt
from IPython.display import display, clear_output

pd.options.display.float_format = '{:,.3f}'.format

# Kafka Consumer Configuration
consumer_config = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "random-user-group",
    "auto.offset.reset": "earliest"
}

consumer = Consumer(consumer_config)
topic_name = 'random-user-topic'
consumer.subscribe([topic_name])

# Initialize gender counts and historical data
gender_counts = {"male": 0, "female": 0}
history = {"Step": [], "male": [], "female": []}
step_counter = 0  # Track steps for plotting

# Function to update the line plot
def update_plot():
    plt.figure(figsize=(8, 5))
    plt.plot(history["Step"], history["male"], marker='o', linestyle='-', label="male", color='blue')
    plt.plot(history["Step"], history["female"], marker='s', linestyle='-', label="female", color='red')

    plt.xlabel("Number of Additions")
    plt.ylabel("Count")
    plt.title("Gender Count Over Time")
    plt.legend()
    plt.grid(True)
    
    clear_output(wait=True)  # Clear previous output to refresh display
    plt.show()

# List to store received data
random_user_data = []

# Kafka Consumer Loop
try:
    while True:
        msg = consumer.poll(timeout=1.0)

        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            else:
                print(f"Kafka Error: {msg.error()}")
                break

        # Parse JSON message from Kafka
        random_user_data_point = json.loads(msg.value().decode("utf-8"))
        random_user_data.append(random_user_data_point)

        # Create DataFrame from collected data
        random_user_df = pd.DataFrame(random_user_data)

        # Check if "Gender" exists before counting
        if "Gender" in random_user_df.columns:
            # Count occurrences of each gender
            gender = random_user_data_point.get("Gender", None)
            if gender in gender_counts:
                gender_counts[gender] += 1
                step_counter += 1  # Increment step count
                
                # Update history for plotting
                history["Step"].append(step_counter)
                history["male"].append(gender_counts["male"])
                history["female"].append(gender_counts["female"])

                # Update the plot
                update_plot()
        
        
        display(random_user_df)
        time.sleep(1)  # Delay to prevent excessive CPU usage

except KeyboardInterrupt:
    print("Stopped by user")

finally:
    consumer.close()  # Safely close the Kafka consumer

Stopped by user

	Title	First Name	Last Name	Gender	Phone
0	Ms	Olivia	Woods	female	021-018-1790
1	Mrs	Ruby	White	female	(764)-202-1525
2	Mr	Koray	Topaloğlu	male	(154)-087-8658
3	Mr	Adam	Johansen	male	40788322
4	Miss	Camila	Johnson	female	08-2096-6197
5	Mr	Liam	Hall	male	(403)-562-3085
6	Mr	Noah	Robinson	male	(048)-480-8057
7	Mr	Augustin	Richard	male	03-78-56-50-51
8	Mr	Barış	Tekand	male	(143)-511-3515

	Title	First Name	Last Name	Gender	Phone
0	Miss	Séléna	Chevalier	female	03-52-65-47-12
1	Madame	Annie	Gautier	female	077 261 55 34
2	Ms	Margot	Sanchez	female	04-79-97-87-58
3	Mr	Sami	Lagesen	male	72407906
4	Mrs	Alexis	Taylor	female	(473)-412-5992
5	Mr	Nathaniel	Brown	male	(949)-969-3801
6	Mr	Eemeli	Puro	male	02-941-346
7	Miss	Hiske	Broeken	female	(008) 4487509
8	Mr	Vedat	Mayhoş	male	(628)-984-5521
9	Mr	Andreas	Christiansen	male	85819190
10	Ms	Laura	Møller	female	90242902
11	Mr	Kevin	Hart	male	(439) 584-1344
12	Mr	Mijat	Srećković	male	030-3797-758
13	Mr	پوریا	رضاییان	male	067-22401238
14	Mr	Joey	Burton	male	013873 56847

Apache Kafka¶

Sumber Data:¶

Persiapan Data:¶

Kafka Producer¶

Kafka Consumer¶

`try`, `except`, dan `finally`¶

2️⃣ `except`¶

3️⃣ `finally`¶

Apache Kafka¶

Sumber Data:¶

Persiapan Data:¶

Kafka Producer¶

Kafka Consumer¶

try, except, dan finally¶

2️⃣ except¶

3️⃣ finally¶

`try`, `except`, dan `finally`¶

2️⃣ `except`¶

3️⃣ `finally`¶