68 lines
1.7 KiB
Python
68 lines
1.7 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
from kafka import KafkaConsumer
|
||
|
|
import boto3
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
from datetime import datetime
|
||
|
|
|
||
|
|
# Configuratie
|
||
|
|
KAFKA_BROKER = "10.0.21.36:9092"
|
||
|
|
KAFKA_TOPIC = "test-lakehouse"
|
||
|
|
S3_ENDPOINT = "http://10.0.20.111:9020"
|
||
|
|
S3_BUCKET = "data"
|
||
|
|
S3_ACCESS_KEY = "REDACTED"
|
||
|
|
S3_SECRET_KEY = "REDACTED"
|
||
|
|
|
||
|
|
print("=" * 60)
|
||
|
|
print(f"Kafka Consumer starting...")
|
||
|
|
print(f"Kafka broker: {KAFKA_BROKER}")
|
||
|
|
print(f"Kafka topic: {KAFKA_TOPIC}")
|
||
|
|
print(f"S3 endpoint: {S3_ENDPOINT}")
|
||
|
|
print(f"S3 bucket: {S3_BUCKET}")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
# S3 client
|
||
|
|
s3 = boto3.client(
|
||
|
|
's3',
|
||
|
|
endpoint_url=S3_ENDPOINT,
|
||
|
|
aws_access_key_id=S3_ACCESS_KEY,
|
||
|
|
aws_secret_access_key=S3_SECRET_KEY,
|
||
|
|
use_ssl=False,
|
||
|
|
verify=False
|
||
|
|
)
|
||
|
|
|
||
|
|
# Kafka consumer
|
||
|
|
try:
|
||
|
|
consumer = KafkaConsumer(
|
||
|
|
KAFKA_TOPIC,
|
||
|
|
bootstrap_servers=[KAFKA_BROKER],
|
||
|
|
auto_offset_reset='earliest',
|
||
|
|
enable_auto_commit=True,
|
||
|
|
value_deserializer=lambda x: json.loads(x.decode('utf-8'))
|
||
|
|
)
|
||
|
|
print("Connected to Kafka successfully!")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Failed to connect to Kafka: {e}")
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
print(f"Listening for messages on topic '{KAFKA_TOPIC}'...")
|
||
|
|
print("-" * 60)
|
||
|
|
|
||
|
|
message_count = 0
|
||
|
|
for message in consumer:
|
||
|
|
message_count += 1
|
||
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S_%f')
|
||
|
|
key = f"kafka/{KAFKA_TOPIC}/message_{timestamp}.json"
|
||
|
|
|
||
|
|
try:
|
||
|
|
s3.put_object(
|
||
|
|
Bucket=S3_BUCKET,
|
||
|
|
Key=key,
|
||
|
|
Body=json.dumps(message.value, indent=2)
|
||
|
|
)
|
||
|
|
print(f"[{message_count}] Written to s3://{S3_BUCKET}/{key}")
|
||
|
|
print(f" Data: {json.dumps(message.value)[:100]}...")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error writing to S3: {e}")
|