Python streaming (spark+kafka)

Привет, меня зовут Роман Вороновский и в этой статье, посвященной Python streaming с использованием Spark и Kafka мы рассмотрим основные шаги, представленные в видео:

Чтобы предоставить вам более подробное описание процесса и помочь при развертывании локального окружения.

Видео,как и статья нацелена на тех, кто только начинает свой путь в Big Data и еще не совсем умеет настраивать различные окружения для легкого старта в профессии.

Итак начнем.

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

Для установки необходимого окружения и проверки компонентов, перед началом работы с Python streaming в связке Spark и Kafka, вам потребуются следующие инструменты и плагины.

0. Docker Desktop (скачать можно по ссылке: https://www.docker.com/products/docker-desktop/)

1. Spark: Apache Spark — это быстрая и мощная платформа для обработки больших объемов данных. Для установки Spark достаточно воспользоваться интерпретатором conda (скачать можно по ссылке: https://www.anaconda.com/download) а не стандартным интерпретатором python, для этого:

• в правом нижнем углу жмем по интерпретаторам

• выбираем опцию «Add new interpreter»

• выбираем опцию «Add local interpreter»

• выбираем Conda Environment

С данным интерпретатором Spark работает из коробки и никаких допов в виде Hadoop и тд. разворачивать не прийдется.

2. Kafka: Apache Kafka — это распределенная платформа, предназначенная для обработки потоковых данных. Для установки Kafka достаточно использовать docker-compose.yaml файл, закрепленный в репозитории. Для его запуска можно воспользоваться командой, находясь в корне проекта:

docker-compose up -d

3. Python: Python — популярный язык программирования, используемый для разработки приложений и анализа данных. Убедитесь, что на вашем компьютере установлена актуальная версия Python. Вы можете скачать и установить Python с официального сайта (https://www.python.org/downloads). В видео я использую IDE- PyCharm (скачать можно по ссылке: https://www.jetbrains.com/pycharm/?var=1)

4. Для работы с Kafka в Python, вам понадобится установить библиотеку pykafka. Откройте командную строку и выполните команду:

pip install pykafka

для установки этой библиотеки.

5.PySpark — это Python API для Apache Spark. Чтобы установить PySpark, выполните команду:

pip install pyspark

в командной строке.

После установки всех необходимых компонентов и плагинов, проверяем их работоспособность.

Проверка работоспособности

Kafka

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

Запускаем Python скрипт который бы писал сообщения в Kafka(producer.py):

from pykafka import KafkaClient  if __name__ == "__main__":     #хост подключения к кафка     client = KafkaClient(hosts='127.0.0.1:9092')     #имя топика в который мы собираемся отправлять сообщения     topic = client.topics[b'stream_topic']     #создание продюсера, который и будет отправлять сообщения     producer = topic.get_producer()     #отправка самих сообщений     producer.produce(b'Hello, Streaming!')     producer.produce(b'Kafka first Message!')     #остановка продюсера     producer.stop()

Следом запускаем скрипт который бы распечатывал сообщения из топика kafka(consumer.py):

from pykafka import KafkaClient  if __name__ == "__main__":     #хост подключения к кафка     client = KafkaClient(hosts='127.0.0.1:9092')     #имя топика из которого мы собираемся получать сообщения     topic = client.topics[b'stream_topic']     #создание простого консьюмера     consumer = topic.get_simple_consumer()     #обработка сообщений если они еще есть в топике     for message in consumer:         if message is not None:             print(message.value.decode('utf-8')) #берется значение сообщений и декодируется по utf-8 для валидного отображения

В результате все сообщения которые были отправлены должны были распечататься в консоль при помощи consumer.py программы.

Spark+kafka

Для проверки уже в связке необходимо использовать программу — sparkConsumeStreamKafkaWriteToConsole.py :

import os  from pyspark.sql import SparkSession os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.3.0,org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 pyspark-shell'  if __name__ == '__main__':     spark = SparkSession \         .builder \         .appName("ProduceConsoleApp") \         .getOrCreate()      source = (spark.readStream.format("kafka")               .option("kafka.bootstrap.servers", "127.0.0.1:9092")               .option("subscribe", "stream_topic")               .option("startingOffsets", "earliest")               .load()               )     #печать схемы - необязательна     source.printSchema()     #работа над получаемыми DataFrame     df = (source           .selectExpr('CAST(value AS STRING)', 'offset'))     # блок записи полученных результатов после действий над DataFrame     console = (df                .writeStream                .format('console'))      console.start().awaitTermination() 

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

В блоке записи мы можем указывать различные форматы, так же для примера был выбран самый простой — в консоль. Если хотите усложнить и прописать обратную запись в какой-нибудь результирующий топик, то попробуйте изменить код начиная с .format следующим образом:

 .format("kafka") \         .option("kafka.bootstrap.servers", "127.0.0.1:9092") \         .option("topic", "result_topic") \         .option('checkpointLocation', './.local/checkpoint') \         .start().awaitTermination()

установка Hadoop_Home

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

Первое с чего стоит начать — установить себе папку winutils которую я показывал на видео(скачать можно по ссылке: https://github.com/kontext-tech/winutils). Я использовал другую версию, которая также закреплена линком в репозитории, привожу её и тут: https://github.com/cdarlint/winutils

После успешной установки(в моем случае это версия 3.2.2) необходимо задать системную переменную HADOOP_HOME. Делается это следующим образом:

• заходим в параметры системы

• прописываем: «изменение системных переменных среды»

• жмем на кнопку: «переменные среды»

• жмем на кнопку: «создать»

• в поле имя переменной прописываем: «HADOOP_HOME»

• в поле значение переменной прописываем путь до нашего скаченного файла нужной версии ( в моем случае: «C:\Users\Roman\Documents\hadoop-3.2.2»

• Находим среди системных переменных- переменную с наименованием: «Path»

• изменяем её дописывая следующий текст: «%HADOOP_HOME%\bin»

• сохраняем и перезагружаем ПК

Листинг изображениями:

Итог

После проделанных манипуляций ошибка уйдет и проект должен работать корректно, в данном случае мои версии для которых все запускалось:

• Python 3.11 — conda interpreter

• pyspark 3.4.1

• pykafka 2.8.0

Всем спасибо за внимание! Оставляйте комментарии и подписывайтесь!

Репозиторий закреплен под видео.