release commit

Author: nikborovets

.gitignore

@@ -0,0 +1,3 @@
+.DS_Store
+
+myenv

README.md

@@ -0,0 +1,39 @@
+# Kafka-Postgres Learning Project
+
+Этот проект создан для изучения работы с Apache Kafka и PostgreSQL с использованием Python.
+
+## Описание
+
+Проект включает три основных компонента:
+
+1. **Producer**: Генерирует данные заказов и отправляет их в Kafka.
+2. **Consumer**: Читает данные заказов из Kafka и сохраняет их в PostgreSQL.
+3. **Analyzer**: Анализирует данные заказов, хранящиеся в PostgreSQL.
+
+## Установка
+
+### Bash
+
+Запустите скрипт `setup.sh`, который автоматически создаст виртуальное окружение, установит необходимые зависимости из `requirements.txt`, настроит докер-контейнеры для Kafka, Zookeeper и PostgreSQL, а также проверит их состояние перед созданием таблицы в базе данных PostgreSQL.
+
+Если Python установлен не в /usr/bin/python3, то измените файл `setup.sh`, заменив /usr/bin/python3 на путь к вашему python3. 
+
+```sh
+./setup.sh
+```
+
+### Важно!
+
+Для наглядности рекомендуется запускать `producer.py`, `consumer.py` и `analyzer.py` в отдельных терминалах.
+
+```sh
+source myenv/bin/activate && python producer.py
+```
+
+```sh
+source myenv/bin/activate && python consumer.py
+```
+
+```sh
+source myenv/bin/activate && python analyzer.py
+```

analyzer.py

@@ -0,0 +1,14 @@
+import psycopg2
+
+conn = psycopg2.connect(
+    dbname='orders_db',
+    user='user',
+    password='password',
+    host='localhost',
+    port=5433
+)
+cursor = conn.cursor()
+
+cursor.execute("SELECT COUNT(*), SUM(amount) FROM orders")
+count, total_amount = cursor.fetchone()
+print(f"Total Orders: {count}, Total Amount: {total_amount}")

check.txt

@@ -0,0 +1,32 @@
+Python 3.9.6
+virtualenv -p /usr/bin/python3 myenv
+source myenv/bin/activate
+
+
+docker-compose up -d
+
+docker exec -it kafka_postgres_learning-postgres-1 psql -U user -d orders_db
+
+CREATE TABLE orders (
+    order_id SERIAL PRIMARY KEY,
+    user_id INT NOT NULL,
+    amount DECIMAL NOT NULL,
+    timestamp BIGINT NOT NULL
+);
+
+source myenv/bin/activate & python producer.py
+python consumer.py
+python analize.py
+
+chmod +x setup.sh
+
+
+
+docker-compose down
+docker-compose ps
+
+DROP TABLE orders;
+
+docker images -a -q
+docker images
+docker rmi $(docker images -a -q)

consumer.py

@@ -0,0 +1,33 @@
+from kafka import KafkaConsumer
+import json
+import psycopg2
+
+consumer = KafkaConsumer(
+    'orders',
+    bootstrap_servers='localhost:9092',
+    auto_offset_reset='earliest',
+    enable_auto_commit=True,
+    group_id='order-group',
+    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
+)
+
+conn = psycopg2.connect(
+    dbname='orders_db',
+    user='user',
+    password='password',
+    host='localhost',
+    port=5433
+)
+cursor = conn.cursor()
+
+def save_order(order):
+    cursor.execute(
+        "INSERT INTO orders (user_id, amount, timestamp) VALUES (%s, %s, %s)",
+        (order['user_id'], order['amount'], order['timestamp'])
+    )
+    conn.commit()
+
+for message in consumer:
+    order = message.value
+    save_order(order)
+    print(f"Consumed: {order}")

docker-compose.yml

@@ -0,0 +1,28 @@
+version: '3.8'
+services:
+  zookeeper:
+    image: bitnami/zookeeper:latest
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+    ports:
+      - "2181:2181"
+
+  kafka:
+    image: bitnami/kafka:latest
+    environment:
+      - KAFKA_BROKER_ID=1
+      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+      - KAFKA_LISTENERS=PLAINTEXT://:9092
+      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
+      - ALLOW_PLAINTEXT_LISTENER=yes
+    ports:
+      - "9092:9092"
+
+  postgres:
+    image: postgres:latest
+    environment:
+      POSTGRES_USER: user
+      POSTGRES_PASSWORD: password
+      POSTGRES_DB: orders_db
+    ports:
+      - "5433:5432"

producer.py

@@ -0,0 +1,23 @@
+from kafka import KafkaProducer
+import json
+import time
+import random
+
+producer = KafkaProducer(
+    bootstrap_servers='localhost:9092',
+    value_serializer=lambda v: json.dumps(v).encode('utf-8')
+)
+
+def generate_order():
+    order = {
+        'user_id': random.randint(1, 100),
+        'amount': round(random.uniform(10.0, 1000.0), 2),
+        'timestamp': int(time.time())
+    }
+    return order
+
+while True:
+    order = generate_order()
+    producer.send('orders', order)
+    print(f"Produced: {order}")
+    time.sleep(1)

requirements.txt

@@ -0,0 +1,5 @@
+kafka-python==2.0.2
+psycopg2-binary==2.9.9
+six==1.16.0
+SQLAlchemy==2.0.30
+typing_extensions==4.12.2

setup.sh

@@ -0,0 +1,35 @@
+#!/bin/bash
+
+set -e
+
+virtualenv -p /usr/bin/python3 myenv
+source myenv/bin/activate
+
+pip install -r requirements.txt
+
+docker-compose up -d
+
+echo "Waiting for containers to start..."
+containers=(kafka_postgres_learning-postgres-1 kafka_postgres_learning-zookeeper-1 kafka_postgres_learning-kafka-1)
+
+for container in "${containers[@]}"; do
+  while [ "$(docker inspect -f '{{.State.Running}}' $container)" != "true" ]; do
+    echo "Waiting for container $container..."
+    sleep 5
+  done
+done
+
+sleep 3
+
+docker exec -i kafka_postgres_learning-postgres-1 psql -U user -d orders_db << EOF
+CREATE TABLE IF NOT EXISTS orders (
+    order_id SERIAL PRIMARY KEY,
+    user_id INT NOT NULL,
+    amount DECIMAL NOT NULL,
+    timestamp BIGINT NOT NULL
+);
+EOF
+
+# python producer.py &
+# python consumer.py &
+# python analyzer.py &