1
import json
2
import time
3
from kafka import KafkaProducer
4
from datetime import datetime
5
import random
6

7
# Kafka Config
8
KAFKA_BROKER = 'localhost:9092'
9
TOPIC = 'user_events_topic'
10

11
# Initialize Kafka Producer
12
producer = KafkaProducer(
13
    bootstrap_servers=[KAFKA_BROKER],
14
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
15
)
16

17
# Generate Random Events
18
def generate_event():
19
    return {
20
        "EventDate": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
21
        "UserID": random.randint(0, 10),
22
        "Action": random.choice(["click", "view", "purchase"]),
23
        "Version": 1
24
    }
25

26
# Produce Events Continuously
27
def produce():
28
    print("Starting Kafka Producer...")
29
    try:
30
        while True:
31
            event = generate_event()
32
            producer.send(TOPIC, value=event)
33
            producer.flush() # NOTEICED: it'll be cache default, so we flush it
34
            print(f"Produced: {event}")
35
            time.sleep(1)  # Send 1 message per second (adjust as needed)
36
    except KeyboardInterrupt:
37
        print("Stopped Producer.")
38
    finally:
39
        producer.close()
40

41
if __name__ == "__main__":
42
    produce()

1
import clickhouse_connect
2

3
# ClickHouse Config
4
CLICKHOUSE_HOST = 'localhost'
5
CLICKHOUSE_PORT = 8123
6

7
def connect_clickhouse():
8
    """connect to ClickHouse"""
9
    try:
10
        client = clickhouse_connect.get_client(
11
            host=CLICKHOUSE_HOST,
12
            port=CLICKHOUSE_PORT,
13
            username='default',
14
            password='default'
15
        )
16
        print("connected to ClickHouse!")
17
        return client
18
    except Exception as e:
19
        print(f"connect to ClickHouse failed: {e}")
20
        return None
21

22
def check_tables(client):
23
    """check tables"""
24
    try:
25
        result = client.query("SHOW TABLES")
26
        print("existing tables:")
27
        for row in result.result_rows:
28
            print(f"  - {row[0]}")
29
        return True
30
    except Exception as e:
31
        print(f"check tables failed: {e}")
32
        return False
33

34
def query_data(client):
35
    """query data"""
36
    try:
37
        # query total records
38
        count_result = client.query("SELECT COUNT(*) FROM default.user_events")
39
        total_count = count_result.result_rows[0][0]
40
        print(f"\ntotal records: {total_count}")
41

42
        if total_count > 0:
43
            # query recent 10 records
44
            recent_result = client.query("""
45
                SELECT EventDate, UserID, Action
46
                FROM default.user_events
47
                ORDER BY EventDate DESC
48
                LIMIT 10
49
            """)
50

51
            print("\nrecent 10 records:")
52
            print("-" * 60)
53
            for row in recent_result.result_rows:
54
                print(f"  {row[0]} | UserID: {row[1]} | Action: {row[2]}")
55

56
        # group by Action
57
        action_stats = client.query("""
58
            SELECT Action, COUNT(*) as count
59
            FROM default.user_events
60
            GROUP BY Action
61
            ORDER BY count DESC
62
        """)
63

64
        print("\ngroup by Action:")
65
        print("-" * 30)
66
        for row in action_stats.result_rows:
67
            print(f"  {row[0]}: {row[1]} records")
68

69
    except Exception as e:
70
        print(f"query data failed: {e}")
71

72
def main():
73
    print("ClickHouse data query tool")
74
    print("=" * 40)
75

76
    client = connect_clickhouse()
77
    if client:
78
        check_tables(client)
79
        query_data(client)
80
        client.close()
81

82
if __name__ == "__main__":
83
    main()

1
-- 查詢最近 10 筆
2
SELECT EventDate, UserID, Action
3
FROM default.user_events
4
ORDER BY EventDate DESC
5
LIMIT 10
6

7
-- Group by 統計
8
SELECT Action, COUNT(*)
9
FROM default.user_events
10
GROUP BY Action
11
ORDER BY count DESC