Полная версия:
Библиотеки Python Часть 2. Практическое применение
consumer = Consumer({
'bootstrap.servers': broker,
'group.id': 'sensor-group',
'auto.offset.reset': 'earliest'
})
consumer.subscribe(['sensor_data'])
# Открываем файл для записи
with open('high_temp.json', 'w') as outfile:
try:
while True:
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
continue
# Преобразуем сообщение в Python-объект
sensor_data = json.loads(msg.value().decode('utf-8'))
# Сохраняем данные, если температура выше 30°C
if sensor_data['temperature'] > 30:
json.dump(sensor_data, outfile)
outfile.write('\n') # Новый ряд для каждого объекта
except KeyboardInterrupt:
print("Завершение работы.")
finally:
consumer.close()
```
Объяснение:
– Консьюмер читает данные из топика `sensor_data`.
– Данные с температурой выше 30°C записываются в файл `high_temp.json`.
Задача 5: Обнаружение аномалий в данных
Описание:
В топик `temperature_readings` поступают данные о температуре из различных городов:
– `city` – название города.
– `temperature` – измеренная температура.
– `timestamp` – время измерения.
Ваша задача: написать программу, которая будет находить и выводить аномалии – случаи, когда температура превышает 40°C или опускается ниже -10°C.
Решение:
```python
from confluent_kafka import Consumer
import json
# Настройки Kafka
broker = 'localhost:9092'
# Создание консьюмера
consumer = Consumer({
'bootstrap.servers': broker,
'group.id': 'temperature-group',
'auto.offset.reset': 'earliest'
})
consumer.subscribe(['temperature_readings'])
try:
while True:
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
continue
# Преобразуем сообщение в Python-объект
reading = json.loads(msg.value().decode('utf-8'))
# Проверяем на аномалии
if reading['temperature'] > 40 or reading['temperature'] < -10:
print(f"Аномалия! Город: {reading['city']}, Температура: {reading['temperature']}°C")
except KeyboardInterrupt:
print("Завершение работы.")
finally:
consumer.close()
```
Объяснение:
– Консьюмер читает данные о температуре из топика.
– Если температура выходит за пределы нормального диапазона, программа выводит сообщение об аномалии.
Задача 6: Потоковое объединение данных
Описание:
Есть два топика:
1. `orders` – содержит данные о заказах: `order_id`, `product_id`, `quantity`.
2. `products` – содержит данные о товарах: `product_id`, `product_name`, `price`.
Ваша задача: написать программу, которая объединяет данные из этих двух топиков и выводит итоговую информацию о каждом заказе, включая название продукта и общую стоимость.
Решение:
```python
from confluent_kafka import Consumer
import json
# Настройки Kafka
broker = 'localhost:9092'
# Создание консьюмеров для обоих топиков
order_consumer = Consumer({
'bootstrap.servers': broker,
'group.id': 'order-group',
'auto.offset.reset': 'earliest'
})
product_consumer = Consumer({
'bootstrap.servers': broker,
'group.id': 'product-group',
'auto.offset.reset': 'earliest'
})
order_consumer.subscribe(['orders'])
product_consumer.subscribe(['products'])
# Словарь для хранения данных о товарах
product_catalog = {}
try:
while True:
# Чтение данных из топика products
product_msg = product_consumer.poll(0.1)
if product_msg and not product_msg.error():
product = json.loads(product_msg.value().decode('utf-8'))
product_catalog[product['product_id']] = {
'name': product['product_name'],
'price': product['price']
}
# Чтение данных из топика orders
order_msg = order_consumer.poll(0.1)
if order_msg and not order_msg.error():
order = json.loads(order_msg.value().decode('utf-8'))
product_id = order['product_id']
# Объединение данных о заказе и товаре
if product_id in product_catalog:
product = product_catalog[product_id]
total_price = order['quantity'] * product['price']
print(f"Заказ {order['order_id']}: {product['name']} x {order['quantity']} = {total_price} $")
else:
print(f"Информация о товаре {product_id} отсутствует.")
except KeyboardInterrupt:
print("Завершение работы.")
finally:
order_consumer.close()
product_consumer.close()
```
Объяснение:
– Данные из топика `products` кэшируются в словаре `product_catalog`.
– При чтении заказа из топика `orders` программа объединяет данные и вычисляет итоговую стоимость.
Задача 7: Потоковая обработка с вычислением скользящего среднего
Описание:
В топик `stock_prices` поступают данные о ценах акций:
– `symbol` – тикер акции.
– `price` – текущая цена.
– `timestamp` – время.
Ваша задача: вычислять скользящее среднее цены акции за последние 5 сообщений для каждого тикера.
Решение:
```python
from confluent_kafka import Consumer
import json
from collections import defaultdict, deque
# Настройки Kafka
broker = 'localhost:9092'
# Создание консьюмера
consumer = Consumer({
'bootstrap.servers': broker,
'group.id': 'stocks-group',
'auto.offset.reset': 'earliest'
})
consumer.subscribe(['stock_prices'])
# Дек для хранения последних цен по тикерам
price_window = defaultdict(lambda: deque(maxlen=5))
try:
while True:
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
continue
# Преобразуем сообщение в Python-объект
stock_data = json.loads(msg.value().decode('utf-8'))
# Добавляем цену в окно
symbol = stock_data['symbol']
price_window[symbol].append(stock_data['price'])
# Вычисляем скользящее среднее
moving_average = sum(price_window[symbol]) / len(price_window[symbol])
print(f"Скользящее среднее для {symbol}: {moving_average:.2f}")
except KeyboardInterrupt:
print("Завершение работы.")
finally:
consumer.close()
```
Объяснение:
– Используется `deque` для хранения последних 5 цен.
– Скользящее среднее вычисляется как сумма значений, делённая на их количество.
Задача 8: Генерация уведомлений
Описание:
В топик `user_actions` поступают данные о действиях пользователей:
– `user_id` – идентификатор пользователя.
– `action` – выполненное действие (например, "login", "purchase").
Напишите программу, которая отслеживает пользователей, выполнивших вход (`login`), но не совершивших покупку (`purchase`) в течение 10 минут, и отправляет уведомление в топик `notifications`.
Решение:
```python
from confluent_kafka import Consumer, Producer
import json
from datetime import datetime, timedelta
# Настройки Kafka
broker = 'localhost:9092'
# Создание консьюмера
consumer = Consumer({
'bootstrap.servers': broker,
'group.id': 'user-actions-group',
'auto.offset.reset': 'earliest'
})
producer = Producer({'bootstrap.servers': broker})
consumer.subscribe(['user_actions'])
# Словарь для отслеживания пользователей
user_login_time = {}
try:
while True:
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
continue
# Преобразуем сообщение в Python-объект
action = json.loads(msg.value().decode('utf-8'))
user_id = action['user_id']
action_type = action['action']
timestamp = datetime.fromisoformat(action['timestamp'])
if action_type == 'login':
user_login_time[user_id] = timestamp
elif action_type == 'purchase' and user_id in user_login_time:
del user_login_time[user_id]
# Проверяем, прошло ли 10 минут
current_time = datetime.now()
for user, login_time in list(user_login_time.items()):
if current_time – login_time > timedelta(minutes=10):
notification = {'user_id': user, 'message': 'Сделайте покупку!'}
producer.produce('notifications', value=json.dumps(notification))
print(f"Уведомление отправлено для пользователя {user}")
del user_login_time[user]
except KeyboardInterrupt:
print("Завершение работы.")
finally:
consumer.close()
```
Объяснение:
– Время входа пользователей сохраняется в словаре.
– Если с момента входа прошло более 10 минут и покупка не совершена, генерируется уведомление.
Эти задачи показывают, как использовать Apache Kafka для решения реальных задач, таких как фильтрация событий, подсчет статистики, агрегация данных и сохранение обработанной информации. Эти примеры помогут вам освоить основные подходы к работе с потоками данных в реальном времени.
1.3 Работа с базами данных: SQLAlchemy и интеграция с Pandas
SQLAlchemy – это мощная библиотека для работы с базами данных в Python. Она предоставляет инструменты для удобного взаимодействия с реляционными базами данных через ORM (Object Relational Mapping) или с использованием чистого SQL.
Pandas же идеально подходит для анализа данных, но иногда данные, которые мы хотим обработать, хранятся в базах данных. Для этого SQLAlchemy и Pandas можно эффективно интегрировать, чтобы выгружать данные из базы, обрабатывать их в Pandas и сохранять обратно.
Установка и подключение
Для начала работы установите библиотеку SQLAlchemy:
```bash
pip install sqlalchemy
```
Если вы используете SQLite, дополнительных действий не требуется. Для других баз данных, таких как PostgreSQL или MySQL, также потребуется установить драйверы, например:
```bash
pip install psycopg2 # Для PostgreSQL
pip install pymysql # Для MySQL
```
Создайте подключение к базе данных с помощью SQLAlchemy. Например, для SQLite это будет выглядеть так:
```python
from sqlalchemy import create_engine
# Создаем подключение к базе данных SQLite
engine = create_engine('sqlite:///example.db', echo=True)
```
Здесь `echo=True` означает, что в консоль будут выводиться SQL-запросы, выполняемые через SQLAlchemy, что полезно для отладки.
Создание таблиц и работа с ORM
SQLAlchemy поддерживает два основных подхода: работа через ORM и использование SQL-запросов напрямую. Рассмотрим оба.
Создадим таблицу для хранения информации о пользователях:
```python
from sqlalchemy import Table, Column, Integer, String, MetaData
# Создаем метаданные
metadata = MetaData()
# Определяем таблицу
users = Table(
'users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String),
Column('age', Integer),
Column('email', String)
)
# Создаем таблицу в базе данных
metadata.create_all(engine)
```
Теперь таблица `users` создана в базе данных.
Для добавления данных используем объект подключения:
```python
from sqlalchemy import insert
# Подключаемся к базе данных
conn = engine.connect()
# Добавляем данные
insert_query = insert(users).values([
{'name': 'Alice', 'age': 25, 'email': 'alice@example.com'},
{'name': 'Bob', 'age': 30, 'email': 'bob@example.com'},
{'name': 'Charlie', 'age': 35, 'email': 'charlie@example.com'}
])
conn.execute(insert_query)
print("Данные добавлены в таблицу.")
```
Чтение данных и интеграция с Pandas
Чтобы выгрузить данные из базы данных в Pandas, SQLAlchemy предоставляет удобный метод. Используем Pandas для выполнения SQL-запроса:
```python
import pandas as pd
# Чтение данных из таблицы users
query = "SELECT * FROM users"
df = pd.read_sql(query, engine)
print(df)
```
Вывод будет выглядеть так:
```
id name age email
0 1 Alice 25 alice@example.com
1 2 Bob 30 bob@example.com
2 3 Charlie 35 charlie@example.com
```
Теперь данные из базы данных доступны в формате DataFrame, и вы можете применять к ним все мощные инструменты анализа, которые предоставляет Pandas.
Обработка данных с использованием Pandas
Допустим, мы хотим найти всех пользователей старше 30 лет и добавить новый столбец с доменом их электронной почты.
```python
# Фильтрация пользователей старше 30 лет
filtered_df = df[df['age'] > 30]
# Добавление нового столбца с доменом электронной почты
filtered_df['email_domain'] = filtered_df['email'].apply(lambda x: x.split('@')[1])
print(filtered_df)
```
Результат будет выглядеть так:
```
id name age email email_domain
2 3 Charlie 35 charlie@example.com example.com
```
Сохранение данных обратно в базу
После обработки данных в Pandas мы можем сохранить их обратно в базу данных. Для этого Pandas предоставляет метод `to_sql`:
```python
# Сохранение отфильтрованных данных в новую таблицу filtered_users
filtered_df.to_sql('filtered_users', engine, if_exists='replace', index=False)
print("Данные сохранены в таблицу filtered_users.")
```
Теперь в базе данных появилась новая таблица `filtered_users`, содержащая обработанные данные.
Работа с ORM
Для более сложных сценариев SQLAlchemy поддерживает ORM, позволяющий работать с таблицами как с Python-классами.
Определим класс для таблицы `users`:
```python
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
age = Column(Integer)
email = Column(String)
# Создаем сессию для работы с ORM
Session = sessionmaker(bind=engine)
session = Session()
# Пример чтения данных через ORM
users = session.query(User).filter(User.age > 30).all()
for user in users:
print(f"Имя: {user.name}, Возраст: {user.age}, Email: {user.email}")
```
Этот подход особенно удобен, если вы предпочитаете объектно-ориентированный стиль работы с базой данных.
Пример: Анализ данных с SQLAlchemy и Pandas
Представьте, что у вас есть база данных с информацией о продажах, и вы хотите найти города, в которых средняя сумма покупок превышает 5000.
1. Создадим таблицу:
```python
sales = Table(
'sales', metadata,
Column('id', Integer, primary_key=True),
Column('city', String),
Column('amount', Integer)
)
metadata.create_all(engine)
# Добавим данные
conn.execute(insert(sales).values([
{'city': 'New York', 'amount': 7000},
{'city': 'Los Angeles', 'amount': 3000},
{'city': 'New York', 'amount': 8000},
{'city': 'Los Angeles', 'amount': 2000},
{'city': 'Chicago', 'amount': 6000}
]))
```
2. Выгрузим данные и найдем среднюю сумму по городам:
```python
# Чтение данных из таблицы sales
query = "SELECT * FROM sales"
sales_df = pd.read_sql(query, engine)
# Вычисление средней суммы по городам
avg_sales = sales_df.groupby('city')['amount'].mean().reset_index()
# Фильтрация городов с средней суммой > 5000
filtered_sales = avg_sales[avg_sales['amount'] > 5000]
print(filtered_sales)
```
Результат:
```
city amount
0 Chicago 6000.0
1 New York 7500.0
```
3. Сохраним результат в таблицу:
```python
filtered_sales.to_sql('high_avg_sales', engine, if_exists='replace', index=False)
```
Теперь обработанные данные сохранены в базе, и вы можете использовать их в дальнейшем.
SQLAlchemy предоставляет мощные возможности для работы с базами данных, а интеграция с Pandas делает обработку данных ещё более удобной и гибкой. Вы можете быстро выгружать данные из базы, анализировать их с помощью Pandas и сохранять обратно, что упрощает создание аналитических решений и автоматизацию работы с данными.
Задачи для практики
Задача 1: Создание базы данных пользователей и извлечение данных
Описание:
Создайте базу данных `users.db` с таблицей `users`, содержащей следующие столбцы:
– `id` – уникальный идентификатор пользователя.
– `name` – имя пользователя.
– `age` – возраст пользователя.
– `email` – электронная почта.
Добавьте в таблицу данные о пяти пользователях и извлеките всех пользователей старше 30 лет.
Решение:
```python
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData
import pandas as pd
# Создаем подключение к базе данных SQLite
engine = create_engine('sqlite:///users.db', echo=False)
metadata = MetaData()
# Определяем таблицу users
users = Table(
'users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String),
Column('age', Integer),
Column('email', String)
)
# Создаем таблицу
metadata.create_all(engine)
# Добавляем данные
with engine.connect() as conn:
conn.execute(users.insert(), [
{'name': 'Alice', 'age': 25, 'email': 'alice@example.com'},
{'name': 'Bob', 'age': 35, 'email': 'bob@example.com'},
{'name': 'Charlie', 'age': 32, 'email': 'charlie@example.com'},
{'name': 'Diana', 'age': 28, 'email': 'diana@example.com'},
{'name': 'Eve', 'age': 40, 'email': 'eve@example.com'}
])
# Извлечение пользователей старше 30 лет
query = "SELECT * FROM users WHERE age > 30"
df = pd.read_sql(query, engine)
print(df)
```
Результат:
```
id name age email
1 2 Bob 35 bob@example.com
2 3 Charlie 32 charlie@example.com
4 5 Eve 40 eve@example.com
```
Задача 2: Подсчет пользователей по возрастным группам
Описание:
Используя базу данных `users.db`, разделите пользователей на две группы: младше 30 лет и 30 лет и старше. Посчитайте количество пользователей в каждой группе.
Решение:
```python
# Чтение данных из таблицы
df = pd.read_sql("SELECT * FROM users", engine)
# Добавление возрастной группы
df['age_group'] = df['age'].apply(lambda x: 'Under 30' if x < 30 else '30 and above')
# Подсчет пользователей по группам
group_counts = df.groupby('age_group')['id'].count().reset_index()
print(group_counts)
```
Результат:
```
age_group id
0 30 and above 3
1 Under 30 2
```
Задача 3: Сохранение агрегированных данных в новую таблицу
Описание:
Сохраните результаты подсчета пользователей по возрастным группам в новую таблицу `age_groups` в базе данных `users.db`.
Решение:
```python
# Сохранение в новую таблицу
group_counts.to_sql('age_groups', engine, if_exists='replace', index=False)
# Проверка сохраненных данных
saved_data = pd.read_sql("SELECT * FROM age_groups", engine)
print(saved_data)
```
Результат:
```
age_group id
0 30 and above 3
1 Under 30 2
```
Задача 4: Поиск наиболее популярных доменов электронной почты
Описание:
Добавьте данные о пользователях с разными адресами электронной почты. Найдите, какие домены (`example.com`, `gmail.com` и т.д.) встречаются чаще всего.
Решение:
```python
# Добавление новых данных
with engine.connect() as conn:
conn.execute(users.insert(), [
{'name': 'Frank', 'age': 29, 'email': 'frank@gmail.com'},
{'name': 'Grace', 'age': 37, 'email': 'grace@gmail.com'},
{'name': 'Helen', 'age': 33, 'email': 'helen@example.com'}
])
# Чтение данных
df = pd.read_sql("SELECT * FROM users", engine)
# Выделение доменов
df['email_domain'] = df['email'].apply(lambda x: x.split('@')[1])
# Подсчет частоты доменов
domain_counts = df['email_domain'].value_counts().reset_index()
domain_counts.columns = ['email_domain', 'count']
print(domain_counts)
```
Результат:
```
email_domain count
0 example.com 5
1 gmail.com 2
```
Задача 5: Создание таблицы продаж и анализ доходов
Описание:
Создайте таблицу `sales`, содержащую данные о продажах:
– `id` – идентификатор продажи.
– `product` – название продукта.
– `price` – цена продукта.
– `quantity` – количество проданных единиц.
Рассчитайте общий доход для каждого продукта и сохраните результаты в новую таблицу `product_revenues`.
Решение:
```python
# Определение таблицы sales
sales = Table(
'sales', metadata,
Column('id', Integer, primary_key=True),
Column('product', String),
Column('price', Integer),
Column('quantity', Integer)
)
metadata.create_all(engine)
# Добавление данных
with engine.connect() as conn:
conn.execute(sales.insert(), [
{'product': 'Laptop', 'price': 1000, 'quantity': 3},
{'product': 'Phone', 'price': 500, 'quantity': 5},
{'product': 'Tablet', 'price': 300, 'quantity': 7}
])
# Чтение данных
sales_df = pd.read_sql("SELECT * FROM sales", engine)
# Расчет общего дохода
sales_df['revenue'] = sales_df['price'] * sales_df['quantity']
revenues = sales_df.groupby('product')['revenue'].sum().reset_index()
# Сохранение в новую таблицу
revenues.to_sql('product_revenues', engine, if_exists='replace', index=False)
# Проверка сохраненных данных
saved_revenues = pd.read_sql("SELECT * FROM product_revenues", engine)
print(saved_revenues)
```
Результат:
```
product revenue
0 Laptop 3000
1 Phone 2500
2 Tablet 2100
```
Задача 6: Фильтрация данных по динамическому запросу
Описание:
Создайте функцию, которая принимает минимальную цену и возвращает список продуктов, стоимость которых выше указанного значения.
Решение:
```python
def filter_products_by_price(min_price):
query = f"SELECT * FROM sales WHERE price > {min_price}"
result_df = pd.read_sql(query, engine)
return result_df
# Фильтрация продуктов с ценой выше 400
filtered_products = filter_products_by_price(400)
print(filtered_products)
```
Результат:
```
id product price quantity
0 1 Laptop 1000 3
1 2 Phone 500 5
```
Задача 7: Определение наиболее активных пользователей
Описание:
В таблице `activity_log` содержатся данные о действиях пользователей:
– `id` – идентификатор записи.
– `user_id` – идентификатор пользователя.
– `action` – выполненное действие.
– `timestamp` – время выполнения действия.
Определите, кто из пользователей совершил наибольшее количество действий.
Решение:
```python
from sqlalchemy import Table, Column, Integer, String, DateTime
from datetime import datetime
# Определение таблицы activity_log
activity_log = Table(
'activity_log', metadata,
Column('id', Integer, primary_key=True),
Column('user_id', Integer),
Column('action', String),
Column('timestamp', DateTime)
)
metadata.create_all(engine)
# Добавление данных
with engine.connect() as conn:
conn.execute(activity_log.insert(), [
{'user_id': 1, 'action': 'login', 'timestamp': datetime(2025, 1, 1, 10, 0)},
{'user_id': 1, 'action': 'purchase', 'timestamp': datetime(2025, 1, 1, 10, 5)},
{'user_id': 2, 'action': 'login', 'timestamp': datetime(2025, 1, 1, 11, 0)},
{'user_id': 1, 'action': 'logout', 'timestamp': datetime(2025, 1, 1, 10, 10)},
{'user_id': 2, 'action': 'purchase', 'timestamp': datetime(2025, 1, 1, 11, 5)},
{'user_id': 2, 'action': 'logout', 'timestamp': datetime(2025, 1, 1, 11, 10)}
])
# Чтение данных
activity_df = pd.read_sql("SELECT * FROM activity_log", engine)
# Подсчет количества действий по пользователям
user_activity = activity_df.groupby('user_id')['id'].count().reset_index()
