Джейд Картер
Библиотеки Python Часть 2. Практическое применение

Слово от автора

Дорогие читатели!

Python – это не просто язык программирования, это универсальный инструмент, который помогает нам решать самые разные задачи, от обработки данных до создания искусственного интеллекта. Во второй части книги я постарался показать, как эти инструменты можно применять в реальных проектах, делая вашу работу не только более эффективной, но и увлекательной.

Каждая глава этой части – это шаг в сторону практики, где мы вместе преодолеваем границы теории и углубляемся в реальные примеры и кейсы. Мне важно было продемонстрировать, что с помощью Python можно не только писать код, но и находить решения там, где это казалось невозможным.

Эта книга – результат моего опыта, наблюдений и экспериментов. Я надеюсь, что она станет для вас не просто руководством, а вдохновением, мотивирующим к изучению новых возможностей. Помните, что любое знание становится ценным, когда его можно применить на практике.

Спасибо за то, что выбрали эту книгу. Пусть она станет вашим верным спутником в мире Python и откроет двери к новым достижениям.

С уважением,

Джейд картер

Глава 1. Работа с большими данными

1.1 Распределенная обработка данных с Dask и PySpark

Работа с большими объемами данных требует инструментов, которые позволяют эффективно распределять вычисления между несколькими процессорами или даже серверами. Python предлагает две мощные библиотеки для таких задач – Dask и PySpark. Каждая из них разработана для обработки больших данных, но они имеют свои уникальные особенности и подходы. Разберем их по отдельности, чтобы понять, как их использовать, и приведем примеры.

Dask: инструмент для масштабирования локальных задач

Dask – это библиотека, которая позволяет расширить вычисления на вашем компьютере, эффективно распределяя их между ядрами процессора или несколькими машинами в кластере. Она идеально подходит для тех случаев, когда объем данных превышает доступную оперативную память, но вы хотите сохранить гибкость работы с Python.

Основные особенности Dask:

1. Dask совместим с большинством популярных библиотек Python, таких как Pandas, NumPy и Scikit-learn.

2. Он поддерживает ленивые вычисления: операции выполняются только при необходимости.

3. Dask позволяет работать как с массивами данных (аналог NumPy), так и с таблицами (аналог Pandas).

Пример использования Dask для обработки данных:

Предположим, у нас есть большой CSV-файл с данными о продажах. Его объем превышает объем оперативной памяти, поэтому обычные инструменты, такие как Pandas, не могут загрузить файл целиком.

```python

import dask.dataframe as dd

# Загрузка большого CSV-файла с помощью Dask

df = dd.read_csv('sales_data_large.csv')

# Выполнение простых операций (например, фильтрация по значению)

filtered_df = df[df['sales'] > 1000]

# Группировка и вычисление суммарных продаж

sales_summary = filtered_df.groupby('region')['sales'].sum()

# Выполнение вычислений (операции "ленивые", пока мы не вызовем .compute())

result = sales_summary.compute()

# Вывод результатов

print(result)

```

Объяснение кода:

1. `dd.read_csv()`: Вместо загрузки всего файла в память, Dask загружает его частями (по "чанкам").

2. Ленивые вычисления: Все операции, такие как фильтрация и группировка, откладываются до вызова `compute()`.

3. Параллельное выполнение: Dask автоматически распределяет работу между всеми доступными ядрами процессора.

Когда использовать Dask:

– Когда ваши данные не помещаются в память.

– Когда вы уже используете библиотеки Python, такие как Pandas или NumPy, и хотите масштабировать их.

– Когда вам нужно быстро настроить распределенные вычисления на одной или нескольких машинах.

PySpark: инструмент для кластерного вычисления

PySpark – это Python-интерфейс для Apache Spark, платформы, разработанной специально для обработки больших данных. Spark работает на кластерах, что позволяет масштабировать вычисления до сотен машин.

PySpark особенно популярен в случаях, когда данные хранятся в распределенных системах, таких как HDFS или Amazon S3.

Основные особенности PySpark:

1. PySpark работает с данными в формате **RDD** (Resilient Distributed Dataset) или DataFrame.

2. Он поддерживает широкий спектр операций, включая трансформации данных, машинное обучение и потоковую обработку.

3. PySpark интегрируется с Hadoop и другими системами для хранения больших данных.

Пример использования PySpark для обработки данных:

Допустим, у нас есть большие данные о транзакциях, хранящиеся в формате CSV, и мы хотим вычислить среднее значение транзакций по каждому клиенту.

```python

from pyspark.sql import SparkSession

# Создаем сессию Spark

spark = SparkSession.builder.appName("TransactionAnalysis").getOrCreate()

# Читаем данные из CSV-файла

df = spark.read.csv('transactions_large.csv', header=True, inferSchema=True)

# Выполняем трансформации данных

# 1. Фильтрация транзакций с нулевой суммой

filtered_df = df.filter(df['amount'] > 0)

# 2. Группировка по клиенту и вычисление среднего значения

average_transactions = filtered_df.groupBy('customer_id').avg('amount')

# Показ результатов

average_transactions.show()

# Останавливаем Spark-сессию

spark.stop()

```

Объяснение кода:

1. Создание SparkSession: Это точка входа для работы с PySpark.

2. `spark.read.csv()`: Загружаем данные в формате DataFrame, который поддерживает SQL-подобные операции.

3. Трансформации: Операции, такие как фильтрация и группировка, выполняются параллельно на всех узлах кластера.

4. Результат: PySpark возвращает распределенные данные, которые можно сохранить или преобразовать.

Когда использовать PySpark:

– Когда вы работаете с кластерами и хотите обрабатывать данные на нескольких машинах.

– Когда данные хранятся в распределенных системах, таких как HDFS или Amazon S3.

– Когда нужно интегрировать обработку данных с экосистемой Hadoop.

Сравнение Dask и PySpark

И Dask, и PySpark являются эффективными инструментами для распределенной обработки данных. Выбор между ними зависит от ваших требований. Если вы работаете с данными, которые не помещаются в оперативную память, но ваши вычисления выполняются на одном компьютере, Dask будет лучшим выбором. Если же вы имеете дело с огромными объемами данных, распределенными по нескольким машинам, то PySpark станет незаменимым инструментом.

Обе библиотеки позволяют решать задачи, которые ранее казались невозможными из-за ограничений памяти или производительности, и они помогут вам эффективно работать с данными любого масштаба.

Задачи для практики

Задачи для Dask

Задача 1: Обработка большого CSV-файла

Описание: У вас есть CSV-файл размером 10 ГБ с данными о продажах. Вам нужно вычислить общую сумму продаж по регионам, но файл слишком большой для работы в Pandas.

Решение:

```python

import dask.dataframe as dd

# Загрузка большого CSV-файла

df = dd.read_csv('sales_data_large.csv')

# Проверка структуры данных

print(df.head()) # Показываем первые строки

# Группировка по регионам и подсчет общей суммы продаж

sales_by_region = df.groupby('region')['sales'].sum()

# Выполнение вычислений

result = sales_by_region.compute()

print(result)

```

Объяснение:

– `dd.read_csv` позволяет загружать файлы большего объема, чем объем оперативной памяти.

– `compute` выполняет ленивые вычисления.

Задача 2: Преобразование данных в формате JSON

Описание: Дан файл в формате JSON, содержащий информацию о транзакциях. Необходимо отфильтровать транзакции с суммой менее 1000 и сохранить отфильтрованные данные в новый CSV-файл.

Решение:

```python

import dask.dataframe as dd

# Загрузка JSON-файла

df = dd.read_json('transactions_large.json')

# Фильтрация данных

filtered_df = df[df['amount'] >= 1000]

# Сохранение результатов в новый CSV-файл

filtered_df.to_csv('filtered_transactions_*.csv', index=False)

print("Данные сохранены в файлы CSV.")

```

Объяснение:

– Dask автоматически разбивает данные на части, сохраняя их в несколько CSV-файлов.

– Фильтрация выполняется параллельно.

Задачи для PySpark

Задача 3: Анализ логов

Описание: Имеется файл логов сервера (формат CSV). Ваша задача – подсчитать количество ошибок (строки с `status = "ERROR"`) и вывести их общее количество.

Решение:

```python

from pyspark.sql import SparkSession

# Создаем сессию Spark

spark = SparkSession.builder.appName("LogAnalysis").getOrCreate()

# Загрузка данных из CSV-файла

df = spark.read.csv('server_logs.csv', header=True, inferSchema=True)

# Фильтрация строк с ошибками

errors = df.filter(df['status'] == 'ERROR')

# Подсчет количества ошибок

error_count = errors.count()

print(f"Количество ошибок: {error_count}")

# Завершаем сессию Spark

spark.stop()

```

Объяснение:

– `filter` позволяет выбрать строки с определенным значением.

– `count` подсчитывает количество строк после фильтрации.

Задача 4: Средняя сумма покупок

Описание: Дан CSV-файл с данными о покупках. Ваша задача – вычислить среднюю сумму покупок для каждого клиента.

Решение:

```python

from pyspark.sql import SparkSession

# Создаем сессию Spark

spark = SparkSession.builder.appName("PurchaseAnalysis").getOrCreate()

# Загрузка данных

df = spark.read.csv('purchases.csv', header=True, inferSchema=True)

# Группировка по клиенту и расчет средней суммы покупок

avg_purchases = df.groupBy('customer_id').avg('purchase_amount')

# Показ результатов

avg_purchases.show()

# Завершаем сессию Spark

spark.stop()

```

Объяснение:

– `groupBy` позволяет сгруппировать данные по столбцу.

– `avg` вычисляет среднее значение для каждой группы.

Задача 5: Сортировка больших данных

Описание: У вас есть файл с информацией о транзакциях. Необходимо отсортировать данные по дате транзакции и сохранить результат в новый файл.

Решение:

```python

from pyspark.sql import SparkSession

# Создаем сессию Spark

spark = SparkSession.builder.appName("SortTransactions").getOrCreate()

# Загрузка данных

df = spark.read.csv('transactions_large.csv', header=True, inferSchema=True)

# Сортировка данных по дате

sorted_df = df.orderBy('transaction_date')

# Сохранение отсортированных данных в новый файл

sorted_df.write.csv('sorted_transactions', header=True, mode='overwrite')

print("Данные отсортированы и сохранены.")

# Завершаем сессию Spark

spark.stop()

```

Объяснение:

– `orderBy` сортирует данные по указанному столбцу.

– `write.csv` сохраняет результат в новом файле.

Эти задачи демонстрируют, как использовать Dask и PySpark для работы с большими объемами данных.

– Dask подходит для локальных задач и интеграции с Python-библиотеками.

– PySpark эффективен для кластерной обработки данных и интеграции с экосистемой Hadoop.

Обе библиотеки упрощают решение задач, которые сложно выполнить традиционными методами из-за ограничений памяти или мощности процессора.

1.2 Потоковая обработка данных с Apache Kafka

Apache Kafka – это мощная платформа для обработки потоков данных в реальном времени. Она широко используется для обработки и анализа событий, поступающих из различных источников, таких как веб-серверы, базы данных, датчики IoT, системы мониторинга и многое другое. Kafka обеспечивает высокую производительность, надежность и масштабируемость, что делает её одним из лучших инструментов для потоковой обработки данных.

В основе Apache Kafka лежат несколько ключевых компонентов:

1. Брокеры – серверы, которые принимают, хранят и доставляют данные.

2. Топики – логические каналы, через которые данные передаются.

3. Продюсеры – приложения или устройства, которые отправляют данные в Kafka.

4. Консьюмеры – приложения, которые получают данные из Kafka.

Kafka организует поток данных в виде последовательностей сообщений. Сообщения записываются в топики и разделяются на партиции, что позволяет обрабатывать данные параллельно.

Пример потока данных

Представим, что у нас есть система интернет-магазина, где Kafka используется для обработки событий, таких как заказы, клики на странице, добавление товаров в корзину и платежи. Каждое из этих событий записывается в топик Kafka. Например, топик `orders` может содержать события, описывающие новые заказы.

Установка и настройка Apache Kafka

Перед началом работы убедитесь, что Kafka установлена. Для локальной работы используйте официальные сборки Kafka с сайта [Apache Kafka](https://kafka.apache.org/).

1. Установите Kafka и запустите ZooKeeper, необходимый для управления брокерами.

2. Запустите Kafka-брокер.

3. Создайте топик с помощью команды:

```bash

bin/kafka-topics.sh –create –topic orders –bootstrap-server localhost:9092 –partitions 3 –replication-factor 1

```

Отправка данных в Kafka

Теперь создадим простого продюсера на Python, который будет отправлять данные в топик `orders`. Для работы с Kafka на Python используется библиотека `confluent-kafka`. Установите её с помощью команды:

```bash

pip install confluent-kafka

```

Пример кода, который отправляет сообщения в топик:

```python

from confluent_kafka import Producer

import json

import time

# Настройки продюсера

producer_config = {

'bootstrap.servers': 'localhost:9092' # Адрес Kafka-брокера

}

# Создание продюсера

producer = Producer(producer_config)

# Функция для обратного вызова при успешной отправке сообщения

def delivery_report(err, msg):

if err is not None:

print(f'Ошибка доставки сообщения: {err}')

else:

print(f'Сообщение отправлено: {msg.topic()} [{msg.partition()}]')

# Отправка данных в Kafka

orders = [

{'order_id': 1, 'product': 'Laptop', 'price': 1000},

{'order_id': 2, 'product': 'Phone', 'price': 500},

{'order_id': 3, 'product': 'Headphones', 'price': 150}

]

for order in orders:

producer.produce(

'orders',

key=str(order['order_id']),

value=json.dumps(order),

callback=delivery_report

)

producer.flush() # Отправка сообщений в брокер

time.sleep(1)

```

В этом примере продюсер отправляет JSON-объекты в топик `orders`. Каждое сообщение содержит данные о заказе.

Чтение данных из Kafka

Теперь создадим консьюмера, который будет читать сообщения из топика `orders`.

```python

from confluent_kafka import Consumer, KafkaException

# Настройки консьюмера

consumer_config = {

'bootstrap.servers': 'localhost:9092',

'group.id': 'order-group', # Группа консьюмеров

'auto.offset.reset': 'earliest' # Начало чтения с первой записи

}

# Создание консьюмера

consumer = Consumer(consumer_config)

# Подписка на топик

consumer.subscribe(['orders'])

# Чтение сообщений из Kafka

try:

while True:

msg = consumer.poll(1.0) # Ожидание сообщения (1 секунда)

if msg is None:

continue

if msg.error():

if msg.error().code() == KafkaException._PARTITION_EOF:

# Конец партиции

continue

else:

print(f"Ошибка: {msg.error()}")

break

# Обработка сообщения

print(f"Получено сообщение: {msg.value().decode('utf-8')}")

except KeyboardInterrupt:

print("Завершение работы…")

finally:

# Закрытие консьюмера

consumer.close()

```

В этом примере консьюмер подключается к Kafka, читает сообщения из топика `orders` и выводит их на экран.

Потоковая обработка данных

Kafka часто используется совместно с платформами потоковой обработки, такими как Apache Spark или Apache Flink, для анализа данных в реальном времени. Однако вы также можете обрабатывать данные прямо в Python.

Например, предположим, что мы хотим обработать события из топика `orders` и рассчитать суммарную стоимость всех заказов:

```python

from confluent_kafka import Consumer

import json

# Настройки консьюмера

consumer_config = {

'bootstrap.servers': 'localhost:9092',

'group.id': 'order-sum-group',

'auto.offset.reset': 'earliest'

}

# Создание консьюмера

consumer = Consumer(consumer_config)

consumer.subscribe(['orders'])

# Суммарная стоимость заказов

total_sales = 0

try:

while True:

msg = consumer.poll(1.0)

if msg is None:

continue

if msg.error():

continue

# Обработка сообщения

order = json.loads(msg.value().decode('utf-8'))

total_sales += order['price']

print(f"Обработан заказ: {order['order_id']}, текущая сумма: {total_sales}")

except KeyboardInterrupt:

print(f"Общая сумма всех заказов: {total_sales}")

finally:

consumer.close()

```

Преимущества использования Kafka

1. Высокая производительность. Kafka поддерживает миллионы событий в секунду благодаря своей архитектуре и использованию партиций.

2. Надежность. Данные хранятся в Kafka до тех пор, пока их не обработают все подписчики.

3. Масштабируемость. Kafka легко масштабируется путем добавления новых брокеров.

4. Универсальность. Kafka поддерживает интеграцию с большинством современных инструментов обработки данных.

Apache Kafka предоставляет мощный набор инструментов для потоковой обработки данных. Используя Python, вы можете легко настроить передачу данных, их обработку и анализ в реальном времени. Это особенно полезно для систем, где требуется высокая производительность и минимальная задержка при обработке больших потоков данных.

Задачи для практики

Задача 1: Фильтрация событий по условию

Описание:

У вас есть топик `clickstream`, содержащий события о кликах на веб-сайте. Каждое событие содержит следующие поля:

– `user_id` – идентификатор пользователя.

– `url` – URL-адрес, на который был клик.

– `timestamp` – время клика.

Ваша задача: создать консьюмера, который будет читать события из Kafka, фильтровать только события с URL-адресами, содержащими слово "product", и сохранять их в новый топик `filtered_clicks`.

Решение:

```python

from confluent_kafka import Producer, Consumer

import json

# Настройки Kafka

broker = 'localhost:9092'

# Создание продюсера для записи в новый топик

producer = Producer({'bootstrap.servers': broker})

def produce_filtered_event(event):

producer.produce('filtered_clicks', value=json.dumps(event))

producer.flush()

# Создание консьюмера для чтения из исходного топика

consumer = Consumer({

'bootstrap.servers': broker,

'group.id': 'clickstream-group',

'auto.offset.reset': 'earliest'

})

consumer.subscribe(['clickstream'])

# Чтение и фильтрация событий

try:

while True:

msg = consumer.poll(1.0)

if msg is None:

continue

if msg.error():

continue

# Преобразуем сообщение из Kafka в Python-объект

event = json.loads(msg.value().decode('utf-8'))

# Фильтруем события с URL, содержащими "product"

if 'product' in event['url']:

print(f"Фильтруем событие: {event}")

produce_filtered_event(event)

except KeyboardInterrupt:

print("Завершение работы.")

finally:

consumer.close()

```

Объяснение:

– Консьюмер читает события из топика `clickstream`.

– Каждое сообщение проверяется на наличие слова "product" в поле `url`.

– Отфильтрованные события отправляются в новый топик `filtered_clicks` через продюсера.

Задача 2: Подсчет количества событий в реальном времени

Описание:

Топик `log_events` содержит логи системы. Каждое сообщение содержит:

– `log_level` (например, "INFO", "ERROR", "DEBUG").

– `message` (текст лога).

Ваша задача: написать программу, которая считает количество событий уровня "ERROR" в реальном времени и каждые 10 секунд выводит их общее количество.

Решение:

```python

from confluent_kafka import Consumer

import time

# Настройки Kafka

broker = 'localhost:9092'

# Создание консьюмера

consumer = Consumer({

'bootstrap.servers': broker,

'group.id': 'log-group',

'auto.offset.reset': 'earliest'

})

consumer.subscribe(['log_events'])

error_count = 0

start_time = time.time()

try:

while True:

msg = consumer.poll(1.0)

if msg is None:

continue

if msg.error():

continue

# Преобразуем сообщение в Python-объект

log_event = json.loads(msg.value().decode('utf-8'))

# Увеличиваем счетчик, если уровень лога "ERROR"

if log_event['log_level'] == 'ERROR':

error_count += 1

# Каждые 10 секунд выводим текущий счетчик

if time.time() – start_time >= 10:

print(f"Количество ошибок за последние 10 секунд: {error_count}")

error_count = 0

start_time = time.time()

except KeyboardInterrupt:

print("Завершение работы.")

finally:

consumer.close()

```

Объяснение:

– Консьюмер читает события из топика `log_events`.

– Если уровень лога "ERROR", увеличивается счетчик `error_count`.

– Каждые 10 секунд программа выводит количество событий "ERROR" и сбрасывает счетчик.

Задача 3: Агрегация данных по группам

Описание:

Топик `transactions` содержит данные о финансовых транзакциях:

– `user_id` – идентификатор пользователя.

– `amount` – сумма транзакции.

Ваша задача: написать программу, которая подсчитывает общую сумму транзакций для каждого пользователя и выводит результаты в реальном времени.

Решение:

```python

from confluent_kafka import Consumer

import json

from collections import defaultdict

# Настройки Kafka

broker = 'localhost:9092'

# Создание консьюмера

consumer = Consumer({

'bootstrap.servers': broker,

'group.id': 'transaction-group',

'auto.offset.reset': 'earliest'

})

consumer.subscribe(['transactions'])

# Словарь для хранения сумм по пользователям

user_totals = defaultdict(float)

try:

while True:

msg = consumer.poll(1.0)

if msg is None:

continue

if msg.error():

continue

# Преобразуем сообщение в Python-объект

transaction = json.loads(msg.value().decode('utf-8'))

# Обновляем сумму для пользователя

user_id = transaction['user_id']

user_totals[user_id] += transaction['amount']

# Вывод текущих сумм

print(f"Текущая сумма транзакций по пользователям: {dict(user_totals)}")

except KeyboardInterrupt:

print("Завершение работы.")

finally:

consumer.close()

```

Объяснение:

– Консьюмер читает данные из топика `transactions`.

– Для каждого пользователя обновляется сумма его транзакций в словаре `user_totals`.

– Программа выводит текущие суммы по всем пользователям.

Задача 4: Сохранение обработанных данных в файл

Описание:

Топик `sensor_data` содержит данные с датчиков IoT:

– `sensor_id` – идентификатор датчика.

– `temperature` – измеренная температура.

– `timestamp` – время измерения.

Ваша задача: написать программу, которая сохраняет все данные о температуре выше 30°C в файл `high_temp.json`.

Решение:

```python

from confluent_kafka import Consumer

import json

# Настройки Kafka

broker = 'localhost:9092'

# Создание консьюмера

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()

user_activity.columns = ['user_id', 'action_count']

# Поиск самого активного пользователя

most_active_user = user_activity.loc[user_activity['action_count'].idxmax()]

print(most_active_user)

```

Результат:

```

user_id 1

action_count 3

```

Задача 8: Подсчет действий по типу

Описание: Для каждого типа действия из таблицы `activity_log` подсчитайте, сколько раз оно выполнялось.

Решение:

```python

# Подсчет количества каждого типа действия

action_counts = activity_df['action'].value_counts().reset_index()

action_counts.columns = ['action', 'count']

print(action_counts)

```

Результат:

```

action count

0 login 2

1 purchase 2

2 logout 2

```

Задача 9: Анализ временных меток

Описание: Определите, в какие часы дня пользователи наиболее активны.

Решение:

```python

# Извлечение часа из временных меток

activity_df['hour'] = activity_df['timestamp'].dt.hour

# Подсчет действий по часам

hourly_activity = activity_df.groupby('hour')['id'].count().reset_index()

hourly_activity.columns = ['hour', 'action_count']

print(hourly_activity)

```

Результат:

```

hour action_count

0 10 3

1 11 3

```

Задача 10: Создание таблицы доходов от пользователей

Описание: Используя таблицу `sales`, определите, сколько дохода принёс каждый пользователь, и сохраните результаты в таблицу `user_revenues`.

Решение:

```python

# Добавление данных о продажах с указанием user_id

with engine.connect() as conn:

conn.execute(sales.insert(), [

{'product': 'Laptop', 'price': 1000, 'quantity': 1, 'user_id': 1},

{'product': 'Phone', 'price': 500, 'quantity': 2, 'user_id': 1},

{'product': 'Tablet', 'price': 300, 'quantity': 3, 'user_id': 2}

])

# Чтение данных из sales

sales_df = pd.read_sql("SELECT * FROM sales", engine)

# Расчёт дохода для каждого пользователя

sales_df['revenue'] = sales_df['price'] * sales_df['quantity']

user_revenues = sales_df.groupby('user_id')['revenue'].sum().reset_index()

# Сохранение в новую таблицу

user_revenues.to_sql('user_revenues', engine, if_exists='replace', index=False)

# Проверка результатов

saved_user_revenues = pd.read_sql("SELECT * FROM user_revenues", engine)

print(saved_user_revenues)

```

Результат:

```

user_id revenue

0 1 2000

1 2 900

```

Задача 11: Поиск последнего действия пользователей

Описание:Для каждого пользователя из таблицы `activity_log` найдите его последнее действие.

Решение:

```python

# Поиск последнего действия

last_actions = activity_df.sort_values('timestamp').groupby('user_id').last().reset_index()

last_actions = last_actions[['user_id', 'action', 'timestamp']]

print(last_actions)

```

Результат:

```

user_id action timestamp

0 1 logout 2025-01-01 10:10:00

1 2 logout 2025-01-01 11:10:00

```

Задача 12: Фильтрация пользователей с высоким доходом

Описание: Используя таблицу `user_revenues`, выберите всех пользователей, чей доход превышает 1500.

Решение:

```python

# Чтение данных из user_revenues

user_revenues = pd.read_sql("SELECT * FROM user_revenues", engine)

# Фильтрация пользователей с доходом > 1500

high_revenue_users = user_revenues[user_revenues['revenue'] > 1500]

print(high_revenue_users)

```

Результат:

```

user_id revenue

0 1 2000

```

Задача 13: Распределение доходов по продуктам

Описание: Определите, какой процент от общего дохода приносит каждый продукт.

Решение:

```python

# Подсчет общего дохода

total_revenue = sales_df['revenue'].sum()

# Расчет процента дохода по продуктам

sales_df['revenue_percent'] = (sales_df['revenue'] / total_revenue) * 100

product_revenue_percent = sales_df.groupby('product')['revenue_percent'].sum().reset_index()

print(product_revenue_percent)

```

Результат:

```

product revenue_percent

0 Laptop 50.793651

1 Phone 25.396825

2 Tablet 23.809524

```

Эти задачи демонстрируют, как SQLAlchemy и Pandas могут использоваться вместе для создания, управления и анализа данных в базах данных. Они покрывают такие аспекты, как фильтрация данных, выполнение группировок и агрегатов, интеграция данных и сохранение результатов. Эти примеры помогут вам освоить основные техники работы с базами данных в Python.

Глава 2. Интерактивная визуализация и аналитика

2.1 Использование Plotly для интерактивных графиков

Plotly – это мощная библиотека для создания интерактивных графиков и визуализации данных. Она поддерживает широкий спектр графиков: линейные, столбчатые, тепловые карты, трехмерные визуализации и многие другие. Основное преимущество Plotly – интерактивность: пользователи могут увеличивать масштаб, перемещаться по графикам, а также взаимодействовать с данными в реальном времени.

Для работы с Plotly необходимо установить библиотеку:

```bash

pip install plotly

```

После установки можно использовать Plotly в сочетании с Pandas, что упрощает построение графиков на основе данных из DataFrame. Далее мы подробно рассмотрим примеры использования Plotly для создания различных типов графиков.

Построение простого линейного графика

Рассмотрим пример, где мы визуализируем изменение температуры в течение недели.

```python

import plotly.graph_objects as go

# Данные

days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']

temperatures = [22, 24, 19, 23, 25, 28, 26]

# Создание графика

fig = go.Figure()

fig.add_trace(go.Scatter(

x=days,

y=temperatures,

mode='lines+markers', # Линии с точками

name='Temperature',

line=dict(color='blue', width=2),

marker=dict(size=8)

))

# Настройка заголовков

fig.update_layout(

title='Изменение температуры за неделю',

xaxis_title='День недели',

yaxis_title='Температура (°C)',

template='plotly_white'

)

fig.show()

```

Объяснение:

1. Мы создаём объект `Figure`, добавляя в него данные с помощью `add_trace`.

2. Используем `Scatter` для отображения данных в виде линии с точками.

3. С помощью `update_layout` задаём заголовок графика и подписываем оси.

4. Метод `fig.show()` открывает интерактивный график в браузере.

Построение столбчатого графика

Теперь создадим столбчатый график, чтобы отобразить продажи по различным категориям товаров.

```python

categories = ['Electronics', 'Clothing', 'Groceries', 'Books', 'Furniture']

sales = [1000, 1500, 700, 1200, 900]

fig = go.Figure()

fig.add_trace(go.Bar(

x=categories,

y=sales,

name='Sales',

marker=dict(color='orange')

))

fig.update_layout(

title='Продажи по категориям товаров',

xaxis_title='Категории',

yaxis_title='Сумма продаж ($)',

template='plotly_dark'

)

fig.show()

```

Особенности:

– Используем `go.Bar` для построения столбчатого графика.

– Цвет столбцов задаётся через параметр `marker`.

Построение комбинированного графика

Иногда нужно совмещать разные типы графиков на одном рисунке. Рассмотрим пример, где на одном графике отображаются продажи в виде столбцов и прибыль в виде линии.

```python

profit = [300, 500, 200, 400, 350]

fig = go.Figure()

fig.add_trace(go.Bar(

x=categories,

y=sales,

name='Sales',

marker=dict(color='blue')

))

fig.add_trace(go.Scatter(

x=categories,

y=profit,

mode='lines+markers',

name='Profit',

line=dict(color='green', width=2)

))

fig.update_layout(

title='Продажи и прибыль по категориям товаров',

xaxis_title='Категории',

yaxis_title='Сумма ($)',

barmode='group',

template='plotly_white'

)

fig.show()

```

Что добавлено:

– Комбинация `Bar` и `Scatter` позволяет визуализировать данные разных типов.

– Параметр `barmode='group'` размещает столбцы по группам, чтобы они не перекрывались.

Построение круговой диаграммы

Для отображения долей в процентах часто используется круговая диаграмма. Например, распределение продаж по категориям.

```python

fig = go.Figure()

fig.add_trace(go.Pie(

labels=categories,

values=sales,

hole=0.3 # Полудонат (дырка в центре)

))

fig.update_layout(

title='Распределение продаж по категориям',

template='plotly_white'

)

fig.show()

```

Особенности:

– Используем `go.Pie` для построения круговой диаграммы.

– Параметр `hole` задаёт размер центральной части, превращая график в "пончиковую" диаграмму.

Построение тепловой карты

Тепловые карты полезны для отображения матриц данных, например, уровня продаж в разных регионах и месяцах.

```python

import numpy as np

regions = ['North', 'South', 'East', 'West']

months = ['January', 'February', 'March', 'April']

sales_data = np.random.randint(100, 1000, size=(4, 4))

fig = go.Figure(data=go.Heatmap(

z=sales_data,

x=months,

y=regions,

colorscale='Viridis' # Цветовая схема

))

fig.update_layout(

title='Уровень продаж по регионам и месяцам',

xaxis_title='Месяцы',

yaxis_title='Регионы'

)

fig.show()

```

Объяснение:

– Используем `go.Heatmap` для отображения данных в виде тепловой карты.

– Параметр `colorscale` задаёт цветовую палитру, визуально усиливая различия между значениями.

Построение трёхмерного графика

Plotly поддерживает трёхмерные визуализации. Например, график, отображающий поверхность функции.

```python

x = np.linspace(-5, 5, 50)

y = np.linspace(-5, 5, 50)

X, Y = np.meshgrid(x, y)

Z = np.sin(np.sqrt(X**2 + Y**2))

fig = go.Figure(data=[go.Surface(z=Z, x=X, y=Y)])

fig.update_layout(

title='3D График поверхности',

scene=dict(

xaxis_title='X',

yaxis_title='Y',

zaxis_title='Z'

)

)

fig.show()

```

Особенности:

– Используем `go.Surface` для построения трёхмерной поверхности.

– Параметры `scene` задают подписи к осям в трёхмерном пространстве.

Интерактивность с помощью виджетов

Plotly позволяет добавлять интерактивные элементы, такие как слайдеры. Например, график, где пользователь может выбирать диапазон времени.

```python

from plotly.subplots import make_subplots

years = ['2020', '2021', '2022', '2023']

values = [500, 700, 800, 600]

fig = make_subplots(rows=1, cols=1)

fig.add_trace(go.Scatter(

x=years,

y=values,

mode='lines+markers',

name='Yearly Data'

))

fig.update_layout(

title='Интерактивный график с выбором диапазона',

xaxis=dict(rangeslider=dict(visible=True)), # Добавляем ползунок

template='plotly_white'

)

fig.show()

```

Интерактивность:

– Ползунок позволяет выбирать диапазон данных на оси X.

– Это полезно для работы с временными рядами.

Plotly – универсальный инструмент для создания интерактивных графиков. Благодаря множеству типов графиков и богатым возможностям настройки, библиотека подходит для самых разнообразных задач: от анализа данных до их визуальной презентации. Используя Plotly, вы можете не только создавать красивые графики, но и предоставлять пользователям возможность активно взаимодействовать с ними.

Задачи для практики

Задача 1: Построение графика изменения температуры

Описание:

Имеется набор данных о температуре за неделю:

– Дни: `['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']`

– Температура: `[15, 17, 20, 22, 19, 18, 16]`

Постройте линейный график, отображающий изменение температуры. Подпишите оси и добавьте интерактивность.

Решение:

```python

import plotly.graph_objects as go

# Данные

days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']

temperatures = [15, 17, 20, 22, 19, 18, 16]

# Построение графика

fig = go.Figure()

fig.add_trace(go.Scatter(

x=days,

y=temperatures,

mode='lines+markers',

name='Temperature',

line=dict(color='blue', width=2),

marker=dict(size=8)

))

# Настройка графика

fig.update_layout(

title='Изменение температуры за неделю',

xaxis_title='Дни недели',

yaxis_title='Температура (°C)',

template='plotly_white'

)

# Показ графика

fig.show()

```

Задача 2: Построение круговой диаграммы

Описание:

Имеется информация о продажах по категориям:

– Категории: `['Electronics', 'Clothing', 'Groceries', 'Books', 'Furniture']`

– Продажи: `[1200, 1500, 800, 600, 900]`

Постройте круговую диаграмму, отображающую доли продаж по категориям.

Решение:

```python

import plotly.graph_objects as go

# Данные

categories = ['Electronics', 'Clothing', 'Groceries', 'Books', 'Furniture']

sales = [1200, 1500, 800, 600, 900]

# Построение круговой диаграммы

fig = go.Figure(data=[go.Pie(

labels=categories,

values=sales,

hole=0.4 # Делает диаграмму "пончиковой"

)])

# Настройка графика

fig.update_layout(

title='Распределение продаж по категориям',

template='plotly_white'

)

# Показ графика

fig.show()

```

Задача 3: Построение столбчатого графика с несколькими категориями

Описание:

Имеется информация о продажах в двух магазинах по категориям товаров:

– Категории: `['Electronics', 'Clothing', 'Groceries', 'Books', 'Furniture']`

– Продажи в магазине A: `[1000, 1400, 800, 500, 700]`

– Продажи в магазине B: `[1200, 1500, 600, 700, 900]`

Постройте группированный столбчатый график для сравнения продаж в двух магазинах.

Решение:

```python

# Данные

categories = ['Electronics', 'Clothing', 'Groceries', 'Books', 'Furniture']

sales_a = [1000, 1400, 800, 500, 700]

sales_b = [1200, 1500, 600, 700, 900]

# Построение графика

fig = go.Figure()

fig.add_trace(go.Bar(

x=categories,

y=sales_a,

name='Store A',

marker=dict(color='blue')

))

fig.add_trace(go.Bar(

x=categories,

y=sales_b,

name='Store B',

marker=dict(color='orange')

))

# Настройка графика

fig.update_layout(

title='Сравнение продаж по категориям в двух магазинах',

xaxis_title='Категории',

yaxis_title='Продажи ($)',

barmode='group',

template='plotly_white'

)

# Показ графика

fig.show()

```

Задача 4: Построение тепловой карты продаж по регионам и месяцам

Описание:

Имеются данные о продажах в четырёх регионах за три месяца:

– Регионы: `['North', 'South', 'East', 'West']`

– Месяцы: `['January', 'February', 'March']`

– Продажи (матрица):

```

[[500, 600, 700],

[400, 500, 600],

[700, 800, 900],

[300, 400, 500]]

```

Постройте тепловую карту, отображающую продажи.

Решение:

```python

import plotly.graph_objects as go

# Данные

regions = ['North', 'South', 'East', 'West']

months = ['January', 'February', 'March']

sales_matrix = [

[500, 600, 700],

[400, 500, 600],

[700, 800, 900],

[300, 400, 500]

]

# Построение тепловой карты

fig = go.Figure(data=go.Heatmap(

z=sales_matrix,

x=months,

y=regions,

colorscale='Viridis' # Цветовая схема

))

# Настройка графика

fig.update_layout(

title='Тепловая карта продаж',

xaxis_title='Месяцы',

yaxis_title='Регионы'

)

# Показ графика

fig.show()

```

Задача 5: Построение 3D-графика поверхности функции

Описание: Построить 3D-график для функции ( z = cos(x^2 + y^2) cdot *sin(x – y) ) на диапазоне (x) и (y) от (-5) до (5) с использованием более высокой сетки и с улучшенной цветовой гаммой.

Решение:

```python

import numpy as np

import plotly.graph_objects as go

# Данные

x = np.linspace(-5, 5, 100) # Увеличение разрешения

y = np.linspace(-5, 5, 100)

X, Y = np.meshgrid(x, y)

Z = np.cos(X**2 + Y**2) * np.sin(X – Y) # Сложная функция

# Построение 3D-графика

fig = go.Figure(data=[go.Surface(z=Z, x=X, y=Y, colorscale='Viridis')]) # Изменение цветовой гаммы

# Настройка графика

fig.update_layout(

title='3D График сложной поверхности',

scene=dict(

xaxis_title='X',

yaxis_title='Y',

zaxis_title='Z'

),

scene_camera=dict(

eye=dict(x=1.5, y=1.5, z=1.5) # Изменение угла обзора

)

)

# Показ графика

fig.show()

```

Задача 6: Анимация изменения температуры по дням недели

Описание:

Имеется информация о температуре за каждый день недели для нескольких городов:

– Дни: `['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']`

– Города: `['New York', 'Los Angeles', 'Chicago']`

– Температуры (матрица):

```

New York: [22, 24, 26, 25, 23, 21, 20]

Los Angeles: [30, 31, 29, 28, 27, 26, 25]

Chicago: [15, 18, 20, 17, 16, 14, 12]

```

Создайте анимацию, показывающую изменение температуры для каждого города.

Решение:

```python

import plotly.graph_objects as go

# Данные

days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']

cities = ['New York', 'Los Angeles', 'Chicago']

temperatures = {

'New York': [22, 24, 26, 25, 23, 21, 20],

'Los Angeles': [30, 31, 29, 28, 27, 26, 25],

'Chicago': [15, 18, 20, 17, 16, 14, 12]

}

# Создание анимации

fig = go.Figure()

for city in cities:

fig.add_trace(go.Scatter(

x=days,

y=temperatures[city],

mode='lines+markers',

name=city

))

# Настройка анимации

frames = [

go.Frame(

data=[

go.Scatter(

x=days[:i],

y=temperatures[city][:i],

mode='lines+markers',

name=city

)

for city in cities

]

)

for i in range(1, len(days) + 1)

]

fig.update(frames=frames)

# Настройка кнопок

fig.update_layout(

updatemenus=[

dict(

type='buttons',

showactive=False,

buttons=[

dict(label='Play', method='animate', args=[None, {'frame': {'duration': 500, 'redraw': True}}]),

dict(label='Pause', method='animate', args=[[None], {'frame': {'duration': 0, 'redraw': False}}])

]

)

]

)

# Оформление графика

fig.update_layout(

title='Изменение температуры по дням недели',

xaxis_title='День недели',

yaxis_title='Температура (°C)',

template='plotly_white'

)

fig.show()

```

Задача 7: Трёхмерная анимация COVID-19

Описание:

Используйте вымышленные данные о росте случаев COVID-19 в трёх странах (`USA`, `India`, `Brazil`) за шесть месяцев:

– Месяцы: `['January', 'February', 'March', 'April', 'May', 'June']`

– Число случаев (матрица):

```

USA: [1000, 2000, 4000, 8000, 15000, 20000]

India: [500, 1500, 3000, 6000, 12000, 18000]

Brazil: [800, 1600, 3200, 6400, 13000, 19000]

```

Создайте трёхмерную анимацию, показывающую рост числа случаев по месяцам.

Решение:

```python

import plotly.graph_objects as go

# Данные

months = ['January', 'February', 'March', 'April', 'May', 'June']

countries = ['USA', 'India', 'Brazil']

cases = {

'USA': [1000, 2000, 4000, 8000, 15000, 20000],

'India': [500, 1500, 3000, 6000, 12000, 18000],

'Brazil': [800, 1600, 3200, 6400, 13000, 19000]

}

# Построение графика

fig = go.Figure()

for month, idx in zip(months, range(len(months))):

fig.add_trace(go.Scatter3d(

x=countries,

y=[month] * len(countries),

z=[cases[country][idx] for country in countries],

mode='markers',

marker=dict(size=10, color=[cases[country][idx] for country in countries], colorscale='Viridis'),

name=month

))

# Оформление графика

fig.update_layout(

title='Трёхмерная анимация роста COVID-19',

scene=dict(

xaxis_title='Страна',

yaxis_title='Месяц',

zaxis_title='Число случаев'

),

template='plotly_dark'

)

fig.show()

```

Задача 8: Тепловая карта с аннотациями

Описание:

Имеется таблица оценки студентов по предметам:

– Студенты: `['Alice', 'Bob', 'Charlie', 'Diana']`

– Предметы: `['Math', 'Physics', 'Chemistry', 'Biology']`

– Оценки (матрица):

```

[[85, 90, 78, 92],

[88, 84, 89, 91],

[76, 85, 83, 88],

[90, 92, 80, 87]]

```

Постройте тепловую карту, добавив аннотации с оценками.

Решение:

```python

import plotly.graph_objects as go

# Данные

students = ['Alice', 'Bob', 'Charlie', 'Diana']

subjects = ['Math', 'Physics', 'Chemistry', 'Biology']

grades = [

[85, 90, 78, 92],

[88, 84, 89, 91],

[76, 85, 83, 88],

[90, 92, 80, 87]

]

# Построение тепловой карты

fig = go.Figure(data=go.Heatmap(

z=grades,

x=subjects,

y=students,

colorscale='Blues',

showscale=True

))

# Добавление аннотаций

for i, row in enumerate(grades):

for j, val in enumerate(row):

fig.add_annotation(

x=subjects[j],

Конец ознакомительного фрагмента.

Текст предоставлен ООО «Литрес».

Прочитайте эту книгу целиком, купив полную легальную версию на Литрес.

Безопасно оплатить книгу можно банковской картой Visa, MasterCard, Maestro, со счета мобильного телефона, с платежного терминала, в салоне МТС или Связной, через PayPal, WebMoney, Яндекс.Деньги, QIWI Кошелек, бонусными картами или другим удобным Вам способом.