Spark – это научный проект зародившийся в калифорнийском университете в 2009 и ставший на текущий момент одним из самых популярных инструментов обработки данных. Его главным автором является румыно-канадский учёный в области информатики Матей Захария (Matei Zaharia). Spark изначально был написан на Scala и затем доработан с помощью Java. Данный инструмент является фрэймворком для распределённой обработки как структурированных, так и слабо структурированных данных. Помимо родных языков Scala, Java, а также R, данный фрэймворк возможно применять и в Python-скриптах.  

Spark позволяет работать с различными источниками и приёмниками: будь то реляционные базы данных (PostgreSQL, Oracle), NoSQL (Cassandra, MongoDB), файлы различных форматов (csv, json, parquet) или топики Kafka. Вдобавок к этому в Spark есть 4 модуля, которые упрощают жизнь не только дата-инженеру, но и дата-сайентисту:

• SQL – позволяет писать SQL-подобные запросы над различными источниками данных;

• Streaming – модуль для работы с потоковыми данными. Именно этому модулю по сути и посвящена статья;

• MLlib – набор библиотек машинного обучения;

• GraphX – модуль распределённой обработки графов.

Одной из основных идеологий Spark является такое понятие как «ленивые вычисления» (lazy evaluation). То есть реальные вычисления над данными начнутся не сразу, а только тогда, когда Spark увидит в своём плане конкретное действие – например, сохранение данных.

Ещё одной его идеологией (или особенностью) является принцип – не данные к вычислениям, а вычисления к данным. Этот принцип эффективно применяется на HDFS кластере, где большой файл, поделенный и реплицированный на несколько частей, хранится на разных нодах кластера. Spark сам распределяет (транслирует) код-вычислений по всем узлам кластера, разбивает его на подзадачи, создает план выполнения и отслеживает результат выполнения. Все вычисления происходят в оперативной памяти. Ниже – изображение типичной схемы работы Spark-приложения.

При выполнении Spark-приложения первым через SparkSession (в ранних версиях SparkContext) запускается центральный координатор, называемый Driver, он общается со всеми Worker’ами (рабочими узлами кластера), запрашивает выделение ресурсов (RAM и CPU). Каждый рабочий узел состоит из одного или нескольких Executor’ов (Исполнителей), которые отвечают за выполнение Task’а (Задачи) – наименьшая единица процесса, конкретное конечное действие с порцией данных, которая соответствует конкретному разделу RDD. Исполнители регистрируются в Driver’е, который всегда содержит всё информацию об исполнителях. RDD (Resilient Distributed Dataset) – это простая, неизменяемая, распределенная коллекция объектов, хранящихся в оперативной памяти.

Надо отметить, что Spark не стоит на месте – его регулярно улучшают и оптимизируют. До Spark 1.6 был только RDD API с не самой удобной манерой написания трансформаций, а именно – в манере MapReduce. Потом появились такие сущности как Dataframe и Dataset, которые позволили работать с данными как c виртуальной таблицей, хранящейся в оперативной памяти. Тем самым стало возможным работать с данными в SQL-манере (ура!). 

В Spark 2.0 появился SparkSession, который стал единой точкой входа для всех операций и данных Spark. Тем самым он заменил предыдущие точки входа, такие как SparkContext, SQLContext, HiveContext, SparkConf и StreamingContext (хотя и продолжает их поддерживать).

Начиная с версии 2.4, Kafka стал встроенным форматом, с которым можно работать с помощью модуля Structured Streaming. В 3 версии Spark произошли
большие изменения в оптимизации: значительно улучшен SQL-движок,
улучшена обработка ошибок Python и исключений PySpark, обрезка динамических
партиций и многое другое. Кому интересны все нововведения Spark 3.0, то рекомендую ознакомиться со статьёй: https://databricks.com/session_na20/deep-dive-into-the-new-features-of-apache-spark-3-0

Библиотека Structured Streaming была представлена в Spark 2.0 в 2016 году как часть проекта Apache Spark и в качестве механизма обработки микропакетного потока. С версии Spark 2.2 он был признан стабильным, что делает Structured Streaming стандартным механизмом потоковой обработки в фреймоврке Spark, а технологию DStreams – устаревшим.

Structured Streaming, так же как и его предшественник Spark Streaming, предназначен для работы с потоковыми источниками данных: TCP-сокеты, файлы (из HDFS, S3, локальных ФС), Kafka, AWS Kinesis и помимо этого SQL и NoSQL базы данных.

Библиотека структурированной потоковой передачи позволяет работать со стандартным инструментарием SQL-запросов, основанном на модуле Spark SQL и API его основных структур данных – Dataframe и Dataset, поддерживаемыми в языках Java, Scala, Python и R.

Устаревший Spark Streaming (также называемый DStream) применяет RDD API и работает по принципу микропакетной обработки потока, при которой потоковые вычисления моделируются как непрерывная серия небольших заданий пакетной обработки в стиле map/reduce.

В новом Structured Streaming’е было внедрена единая унифицированная модель программирования и интерфейс для пакетной и потоковой обработки, тем самым отойдя от RDD API. Поток данных стал рассматриваться как неограниченная таблица.

В Structured Streaming есть три режима вывода:

• Append mode
  Только новые строки, добавленные к таблице результатов с момента последнего триггера, будут записаны во внешнее хранилище. Это применимо только в запросах, в которых существующие строки в результирующей таблице (в DataFramе) не могут измениться;

• Update mode
  Во внешнем хранилище будут изменены только те строки, которые были обновлены в таблице результатов с момента последнего триггера. Этот режим работает для приемников данных, которые можно обновлять изнутри, например, для таблицы PostgreSQL;

• Complete mode
  Вся обновленная таблица (DataFrame) будет записана во внешнее хранилище.

Помимо этого, в Structured Streaming есть две очень полезные фичи – window() и withWatermark(). Что это такое:

• window() – позволяет агрегировать данные с разбивкой по временным окнам (отрезкам), например, каждые 30 секунд или каждые 5 минут. Указываем – по какому полю типа timestamp (дата-время, миллисекунды не обязательно) выполняем агрегацию микропакетов и агрегируем. Вдобавок можно задать время скольжения этого окна, например, в 1 минуту. Таким образом, каждую 1 минуту приложение будет выполнять агрегацию за последние 5 минут;

• withWatermark() – предназначен для управления временем ожидания данных для заданного временного периода. Иными словами, позволяет определять «просроченность» приходящих микропакетов. Например, если на часах 12:15, а мы установили окно в 5 минут, такой же период ожидания, и в эти 12:15 пришёл пакет из прошлого за 12:03, то такую порцию данных мы отбросим. Причём это можно делать как по временной метке приёма сообщений, так и по времени их создания на источники (зависит от задачи).

Также важно отметить один очень важный механизм в работе с потоковыми данными – механизм контрольных точек (checkpoint). Их основная цель– обеспечить отказоустойчивость потоковых заданий. Благодаря метаданным, хранящимся в файлах контрольных точек, можно перезапустить обработку в случае сбоя бизнес-логики или технической ошибки. Контрольная точка в Spark представляет из себя физический каталог, отвечающий за хранение четырех типов данных:

• source – файлы в этом каталоге содержат информацию о различных источниках, используемых в потоковом запросе. Например, для Apache Kafka исходный файл с контрольной точкой будет содержать карту между разделами (партициями) топика и смещениями при первом выполнении запроса. Это значение неизменяемо и не меняется при выполнении запроса;

• offsets – содержит файл с информацией о данных, которые будут обрабатываться при выполнении данного микропакета. Он генерируется перед физическим выполнением микропакета и представлен классом apache.spark.sql.execution.streaming.OffsetSeqLog. Последний элемент из файла смещения – это строка JSON с сопоставлением разделов и смещений для каждого источника данных, задействованного в запросе. Например, в случае с источником «Kafka» он будет хранить имя топика, номер партиции и начальное смещение для данного микропакета: {"some_data": {"0": 341}};

• commits (коммиты или логи фиксации) – файл-маркер с информацией о водяном знаке (watermark), который будет использоваться в следующем микропакете. Он представлен классом apache.spark.sql.execution.streaming.CommitLog, а метаданные – org.apache.spark.sql.execution.streaming.CommitMetadata. Файл коммитов содержит только одну запись с водяным знаком, примененным к следующему выполнению запроса: {“nextBatchWatermarkMs”: 1655500250234};

• state (состояние) – местоположение контрольной точки. Также отвечает за хранение состояния, созданного логикой обработки с отслеживанием состояния. Эти файлы хранятся в сжатом виде.

Apache Kafka – это успешный проект компании LinkedIn, который выпущена в свет под лицензией Apache в 2012 году. Это распределённый программный брокер сообщений, т.е. шина сообщений с высокой пропускной способностью, на которой можно в реальном времени обрабатывать все проходящие через неё данные. Если кратко, то Kafka – это распределённый журнал коммитов. У данного инструмента удобный API, он легко масштабируем и отказоустойчив. Базовая терминология:

• producer – генератор/источник сообщений;

• consumer – подписчик/читатель сообщений;

• topic – тема, т.е. тематически отдельный (в идеале однородный) поток сообщений с уникальным наименованием. К примеру: на кабельном телевидении есть канал про рыбалку и, соответственно, все передачи на нём будут только про рыбалку, никак не про кулинарию;

• partition – секция/раздел внутри темы. Бытовой пример: перенесёмся во времена, когда бумажные газеты были  популярным источником информации. Каждое утро у киоска (брокера) выстраивается длинная очередь, чтобы купить одну и туже финансовую (топик) газету (сообщение), и далее каждый уже читает интересную ему информацию (поле из сообщения). Очередь гудит из-за недовольства долгим ожиданием, некоторые опаздывают на работу. Однажды ставят ещё два киоска (партиции), продающие такие же газеты – очередь значительно сокращается;

• offset – смещение.  Упрощённо – это порядковый номер сообщения внутри топика (темы).

Принцип работы:

Источники данных (producer'ы) регулярно отправляют сообщения в Kafka и далее они считываются и обрабатываются различными приложениями, то есть потребителями. Сообщения сохраняются в топике, a потребители подписываются на него для получения новых сообщений. Топики (темы) могут «раздуваться» из-за большого количества поступающих данных,  поэтому предусмотрена возможность делить такие топики на более мелкие секции (partitions) для улучшения производительности и масштабируемости.

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

Напоследок стоит добавить, что Kafka разворачивается и работает вместе с Zookeeper – службой поддержки информации о конфигурации, именовании, обеспечении синхронизации распределенных приложений. Kafka применяет ZooKeeper для хранения метаданных о партициях своих топиков и брокерах, а также для выбора брокера в качестве контроллера. В версии Kafka 2.8.0 появилась возможность работать без Zookeeper: https://itchef.ru/articles/60157/

sudo apt install python3-pip

pip3 install jupyter

jupyter notebook

[shutdown]

sudo apt-get update

sudo apt-get install default-jre

java -version

sudo apt-get install scala

scala -version

pip3 install py4j

>>sudo apt-get install default-jre
переместить в Home

>> скачать spark-3.1.1-bin-hadoop3.2.tgz
>> разорхивировать
sudo tar -zxvf spark-3.1.1-bin-hadoop3.2.tgz

export SPARK_HOME='/home/andrey/spark-3.1.1-bin-hadoop3.2'

export PATH=$SPARK_HOME/bin:$PATH

export PYTHONPATH=$SPARK_HOME/python:$PYTHONPATH

nano ~/.bashrc
[copy all comands creating os variables]
[# - optional]

sudo chmod 777 spark-3.1.1-bin-hadoop3.2
sudo chmod -R 777 spark-3.1.1-bin-hadoop3.2

cd spark-3.1.1-bin-hadoop3.2

cd python

python3

>>> import pyspark
>>> quit()

cd spark-3.1.1-bin-hadoop3.2
sudo chmod 777 python

cd python
sudo chmod 777 pyspark

#download package information from all configured sources
sudo apt-get update

#install java
sudo apt-get install default-jre
java -version

#download binary file
wget -P kafka_distr/ "https://dlcdn.apache.org/kafka/3.0.0/kafka_2.12-3.0.0.tgz"

#extract file
tar -xvzf kafka_distr/kafka_2.12-3.0.0.tgz

#create link on path with kafka
ln -s kafka_distr/kafka_2.12-3.0.0.tgz kafka

#edit config
via~/kafka/config/server.propeties
delete.topic.enable = true

sudo apt update
sudo apt install postgresql postgresql-contrib

#create new role
sudo -u postgres createuser --interactive
>>
	Output
	Enter name of role to add: developer
	Shall the new role be a superuser? (y/n) y
<<

#create db
sudo -u postgres createdb advertising

#add pg-user to Ubuntu
sudo adduser developer
>> 
	enter password: 123456
	Retype new password:
	Full name []: developer
	Room Number []: 1
	Work Phone []: 1
	Home Phone []: 1
	Other []: 1
	Is the information correct? [Y/n] Y
<<


#change password in PostgreSQL for users: postgres and developer
sudo -i -u postgres
psql
ALTER USER postgres PASSWORD 'xxxxxxx';
ALTER USER developer PASSWORD 'xxxxxxx';

>>quite

#connect to db 
sudo -i -u developer
psql -U developer -d advertising

CREATE TABLE ad_country_aggregations (
	row_timestamp	TIMESTAMP	default current_timestamp,
	row_id	SERIAL PRIMARY KEY,
	country VARCHAR(50),
	count	INT,
	min	INT,
	max	INT,
	sum	INT,
	avg	DOUBLE PRECISION,
	time_interval	VARCHAR(50)
);

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/test_1"
kafka_servers = "localhost:9092"
topic = "testTopic"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession

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

adViewsDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "earliest") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

query = adViewsDF.writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .start() \
            .awaitTermination()

Рассмотрим случай, когда необходимо применять Spark ниже 3 версии, но выше 2.3 (если вы используете на проекте версию ещё ниже, то лучше примените не технические навыки, а дипломатические и убедите руководство перейти на более свежую версию). Именно с версии 2.4 Kafka можно применять как встроенный тип источника и приёмника. Держим это в уме и запускаем простое PySpark-приложение для проверки из выше упомянутого примера, без дополнительных подключений (но с py-модулем KafkaUtils).

Примечание:

Чтобы в Jupyter notebook работать со Spark 2.4, достаточно скачать соответствующий jar-ник «spark-2.4.4-bin-hadoop2.7.tgz» и подключить его в приложение.

Приложения для чтения из Kafka:

import findspark
findspark.init('/home/andrey/spark-2.4.4-bin-hadoop2.7')

from pyspark.sql import SparkSession
from pyspark.streaming.kafka import KafkaUtils 

checkpoint_path = "./checkpoint/test_2"
kafka_servers = "localhost:9092"
topic = "testTopic"

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

adViewsDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "earliest") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

query = adViewsDF.writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .start() \
            .awaitTermination()

При запуске скрипта сразу получаем ошибку:

TypeError: an integer is required (got type bytes)

Из топиков на форумах понимаем, что ошибка намекает на несовместимость версии Python 3.8 (и 3.9) для PySpark 2.4. Надо установить Python 3.7 и указать его как основную версию.

Скачиваем и устанавливаем Python 3.7.

sudo apt update
sudo apt install python3.7

Меняем основную версию Python:

Способ 1

nano ~/.bashrc
>> [write OS variable]
alias python=/usr/bin/python3.7

Способ 2

sudo update-alternative --install usr/bin/python python usr/bin/python3.7 1
sudo update-alternative --install usr/bin/python python usr/bin/python3.8 2

Но увы, в случае с Jupyter’ом этого оказалось мало. В нём продолжала работать по умолчанию версия Python 3.8 и самый простой способ поменять её – явно указать в начале скрипта. Также можно воспользоваться модулем venv (который мы и применяли на проекте), но в рамках тренировок в Jupyter’е можно прибегнуть к более короткому способу. Помимо этого, мне пришлось скопировать набор библиотек для Python 3.7 из директории Python 3.8:

# копируем python библиотеки для Jupyter из python3.8 (директория site-packages) в python3.7
cp -R /home/andrey/.local/lib/python3.8/site-packages /home/andrey/.local/lib/python3.7/site-packages
sudo chmod -R 777 /home/andrey/.local/lib/python3.7/site-packages
# Первой строкой в notebook’е Jupiter указывайте явно применение pythhon 3.7
%%script /usr/bin/python3.7

Запускаем приложение ещё раз и получаем новую ошибку:

py4j.protocol.Py4JJavaError: An error occurred while calling o31.load.
: org.apache.spark.sql.AnalysisException: Failed to find data source: kafka. Please deploy the application as per the deployment section of "Structured Streaming + Kafka Integration Guide".;

Идём искать причины и решения. Находим то, что необходимо скачать и подключить в config’e spark-приложения пару jar-ников для Kafka. Помимо этого нужно будет подключить библиотеку «pyspark.streaming.kafka».

.config("spark.jars","/home/andrey/spark_jars/spark-sql-kafka-0-10_2.11-2.4.4.jar,/home/andrey/spark_jars/kafka-clients-0.10.1.0.jar")

Теперь всё готово для запуска тестового приложения.

%%script /usr/bin/python3.7

checkpoint_path = "./checkpoint/test_2"
kafka_servers = "localhost:9092"
topic = "testTopic"

import findspark
findspark.init('/home/andrey/spark-2.4.4-bin-hadoop2.7')

from pyspark.sql import SparkSession
from pyspark.streaming.kafka import KafkaUtils 

spark = (SparkSession
        .builder
        .appName("consumer_structured_streaming_test_2")
	 .config("spark.jars","/home/andrey/spark_jars/spark-sql-kafka-0-10_2.11-2.4.4.jar,/home/andrey/spark_jars/kafka-clients-0.10.1.0.jar")  
        .getOrCreate())

adViewsDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "earliest") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

query = adViewsDF.writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .start() \
            .awaitTermination()

from kafka import KafkaProducer    
from json import dumps
from kafka.errors import KafkaError
from time import sleep
from numpy.random import choice, randint
import time
import datetime

def getRandomValue():
    new_dict = {}
    
    country_list = ['ISL','DEU','SWE','ESP', 'UGA','JPN']
    site_list = ['site1.ru','site2.com','site3.com','site4.es','site5.de','site6.jp',
                 'site7.ru','site8.com','site9.com','site10.es','site11.de','site12.jp']
    ts = time.time() + randint(-15, 0)
    event_time = datetime.datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S') 
    
    new_dict['event_time'] = event_time
    new_dict['ad_id'] = randint(1, 5)
    new_dict['country'] = choice(country_list)
    new_dict['site'] = choice(site_list)
    new_dict['viewing_duration'] = randint(1, 15)
    
    return new_dict

producer = KafkaProducer(bootstrap_servers=['localhost:9092'], 
                         value_serializer = lambda x:dumps(x).encode('utf-8'),
                         compression_type = 'gzip')
topic_name = 'adViews'

k = 0
while k < 60:
    k += 1
    random_data = getRandomValue()
    print(random_data)
    try:
        future = producer.send(topic = topic_name, value = random_data)
        message_metadata = future.get(timeout=10)

        print(f"""[+] message_metadata: {message_metadata.topic}, partition: {message_metadata.partition}, offset: {message_metadata.offset}""")
    except Exception as e:
        print(f">>> Error: {e}")
    finally:
        producer.flush()
    sleep(1)

producer.close()

from kafka import KafkaProducer    
from json import dumps
from kafka.errors import KafkaError
from time import sleep
from numpy.random import choice, randint
import time
import datetime

def getRandomValue():
    adViews_dict = {}
    adClicks_dict = {}
    
    country_list = ['ISL','DEU','SWE','ESP','UGA','JPN','BRA']
    site_list = ['site1.ru','site2.com','site3.com','site4.es','site5.de','site6.jp',
                 'site7.ru','site8.com','site9.com','site10.es','site11.de','site12.jp']
    ts = time.time() + randint(-10, 0)
    event_time = datetime.datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S') 
    
    ts = time.time() + randint(-20, 0)
    click_time = datetime.datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S') 
    
    ad_id = randint(1, 5)
    country = choice(country_list)
    site = choice(site_list)
    viewing_duration = randint(1, 15)
    
    adViews_dict['event_time'] = event_time
    adViews_dict['ad_id'] = ad_id
    adViews_dict['country'] = country
    adViews_dict['site'] = site
    adViews_dict['viewing_duration'] = viewing_duration
    
    adClicks_dict['click_time'] = click_time
    adClicks_dict['ad_id'] = ad_id
    adClicks_dict['country'] = country
    adClicks_dict['site'] = site
    
    return adViews_dict, adClicks_dict

producer1 = KafkaProducer(bootstrap_servers=['localhost:9092'], 
                         value_serializer = lambda x:dumps(x).encode('utf-8'),
                         compression_type = 'gzip')

producer2 = KafkaProducer(bootstrap_servers=['localhost:9092'], 
                         value_serializer = lambda x:dumps(x).encode('utf-8'),
                         compression_type = 'gzip')

k = 0
while k < 60:
    k += 1
    random_data_views, random_data_clicks = getRandomValue()
    print(random_data_views)
    try:
        res = producer1.send(topic = 'adViews', value = random_data_views)
        msg = res.get(timeout=10)
        print(f""" [+] message_metadata: {msg.topic}, partition: {msg.partition}, offset: {msg.offset}""")
    except Exception as e:
        print(f">>> Error: {e}")
    finally:
        producer.flush()
    
    click_is_happend = randint(0, 10)
    if click_is_happend > 5:
        print("+ click:")
        print(random_data_clicks)
        try:
            res = producer2.send(topic = 'adClicks', value = random_data_clicks)
            msg = res.get(timeout=10)
            print(f""" [+] message_metadata: {msg.topic}, partition: {msg.partition}, offset: {msg.offset}""")
        except Exception as e:
            print(f">>> Error: {e}")
        finally:
            producer.flush()
    print("-------------------------------------------------------------------------------------------------")
    sleep(1)

producer1.close()
producer2.close()

ad_id	category
1	clothes
2	games
3	electronics
4	cars
5	travel
6	books

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_1_1"
kafka_servers = "localhost:9092"
topic = "adViews"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession

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

adViewsDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "earliest") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

query = adViewsDF.writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .start() \
            .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_1_2"
kafka_servers = "localhost:9092"
topic = "adViews"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

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

kafkaDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "earliest") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schema = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

query = kafkaDF.select(from_json(col("value"), schema).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
            .writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .trigger(processingTime='5 seconds') \
            .start() \
            .awaitTermination()

import os
import shutil
import glob

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_1_3"
kafka_servers = "localhost:9092"
topic = "adViews"

def remove_thing(path):
    if os.path.isdir(path):
        shutil.rmtree(path)
    else:
        os.remove(path)

def empty_directory(path):
    for i in glob.glob(os.path.join(path, '*')):
        remove_thing(i)

try:
    empty_directory(checkpoint_path)
except Exception as e:
    print("Exception: ", e)

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *


kafka_servers = "localhost:9092"
topic = "adViews"

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

kafkaDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "earliest") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schema = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

query = kafkaDF.select(from_json(col("value"), schema).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
            .writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .trigger(processingTime='5 seconds') \
            .start() \
            .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_1_4"
kafka_servers = "localhost:9092"
topic = "adViews"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

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

kafkaDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "earliest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schema = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

query = kafkaDF.select(from_json(col("value"), schema).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
            .writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .trigger(processingTime='5 seconds') \
            .start() \
            .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_1_6"
kafka_servers = "localhost:9092"
topic = "adViews"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

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

kafkaDF = spark \
            .read \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", """{"adViews":{"1":910,"0":908}}""") \
            .option("endingOffsets", """{"adViews":{"1":928,"0":925}}""") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schema = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

batchDataDF = kafkaDF.select(from_json(col("value"), schema).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
            
batchDataDF.show()

batchDataDF.write.format("json") \
            .mode("overwrite") \
            .save("./results/ex_1_6")

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_1_5"
kafka_servers = "localhost:9092"
topic = "adViews"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

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

kafkaDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schema = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

query = kafkaDF.select(from_json(col("value"), schema).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
            .writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .trigger(processingTime='5 seconds') \
            .start() \
            .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_2_1"
kafka_servers = "localhost:9092"
topic = "adViews"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *


kafka_servers = "localhost:9092"
topic = "adViews"

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

kafkaDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schema = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

query = kafkaDF.select(from_json(col("value"), schema).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
            .filter("t.country = 'UGA'") \
            .writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .trigger(processingTime='5 seconds') \
            .start() \
            .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_2_2"
kafka_servers = "localhost:9092"
topic = "adViews"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *


kafka_servers = "localhost:9092"
topic = "adViews"

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

kafkaDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schema = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

query = kafkaDF.select(from_json(col("value"), schema).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
            .filter("t.country in ('DEU','SWE','ESP') and t.viewing_duration > 4") \
            .writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .trigger(processingTime='5 seconds') \
            .start() \
            .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_3_1"
kafka_servers = "localhost:9092"
topic = "adViews"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

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

schemaAdParam = "ad_id int, category string"

adParametersDF = spark \
                .read \
                .format("csv") \
                .option("header", "true") \
                .option("delimiter","\t") \
                .schema(schemaAdParam) \
                .load("./files/ad_parameters.csv")

#adParametersDF.show()
#adParametersDF.printSchema()

adViewsDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schemaAdViews = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

query = adViewsDF.select(from_json(col("value"), schemaAdViews).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
            .join(adParametersDF, "ad_id") \
            .select("event_time", "ad_id", "category", "country", "site", "viewing_duration") \
            .writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .trigger(processingTime='5 seconds') \
            .start() \
            .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_3_2"
kafka_servers = "localhost:9092"
topic = "adViews"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

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

schemaAdParam = "ad_id int, category string"

adParametersDF = spark \
                .read \
                .format("csv") \
                .option("header", "true") \
                .option("delimiter","\t") \
                .schema(schemaAdParam) \
                .load("./files/ad_parameters.csv")

#adParametersDF.show()
#adParametersDF.printSchema()

adViewsDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schemaAdViews = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

query = adViewsDF.select(from_json(col("value"), schemaAdViews).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
            .join(adParametersDF, "ad_id") \
            .select("event_time", "ad_id", "category", "country", "site", "viewing_duration") \
            .filter("category = 'clothes'") \
            .writeStream.format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .trigger(processingTime='5 seconds') \
            .start() \
            .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_4_1"
kafka_servers = "localhost:9092"
import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_4_2"
kafka_servers = "localhost:9092"
topic = "adViews"
db_host = "localhost:5432/advertising"
db_user = "developer"
db_password = "123456"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

def writeToPostgres(df, epoch_id):  
    intervalDF = df.agg(min("event_time").alias("min_event_time"),
                           max("event_time").alias("max_event_time"))
    min_event_time = intervalDF.head(1)[0][0]
    max_event_time = intervalDF.head(1)[0][1]
    agrDF = df.groupBy("country") \
            .agg(count("*").alias("count"),
                 min("viewing_duration").alias("min"),
                 max("viewing_duration").alias("max"),
                 sum("viewing_duration").alias("sum"),
                 avg("viewing_duration").alias("avg")) \
            .select("country","count", "min", "max", "sum", "avg", 
                   concat(lit(min_event_time), lit(" - "), lit(max_event_time)).alias("time_interval"))
    agrDF.show(truncate=False)
    agrDF.write \
        .format('jdbc') \
        .option("url", "jdbc:postgresql://" + db_host) \
        .option("driver", "org.postgresql.Driver") \
        .option("dbtable", "ad_country_aggregations") \
        .option("user", db_user) \
        .option("password", db_password) \
        .mode("append") \
        .save()

spark = (SparkSession
        .builder
        .appName("consumer_structured_streaming_ex_4_2")
        .config("spark.jars","/home/andrey/spark_jars/postgresql-42.4.0.jar")
        .getOrCreate())

adViewsDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schemaAdViews = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

streamData = adViewsDF.select(from_json(col("value"), schemaAdViews).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \

streamData.writeStream \
    .foreachBatch(writeToPostgres) \
    .option("checkpointLocation", checkpoint_path) \
    .trigger(processingTime='10 seconds') \
    .start() \
    .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_5_1"
kafka_servers = "localhost:9092"
topic = "adViews"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

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

adViewsDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schemaAdViews = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

query = adViewsDF.select(from_json(col("value"), schemaAdViews).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
            .withWatermark("event_time", "10 seconds") \
            .groupBy("country", window("event_time", "10 seconds").alias("timewindow")) \
            .agg(count("*").alias("count"),
                 min("viewing_duration").alias("min"),
                 max("viewing_duration").alias("max"),
                 sum("viewing_duration").alias("sum"),
                 avg("viewing_duration").alias("avg")) \
            .writeStream \
            .format("console") \
            .option("truncate", "false") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .start() \
            .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_5_3"
kafka_servers = "localhost:9092"
topic = "adViews"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

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

adViewsDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe",topic) \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schemaAdViews = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

query = adViewsDF.select(from_json(col("value"), schemaAdViews).alias("t")) \
            .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
            .withWatermark("event_time", "10 seconds") \
            .groupBy("country", window("event_time", "10 seconds", "5 seconds").alias("timewindow")) \
            .agg(count("*").alias("count"),
                 min("viewing_duration").alias("min"),
                 max("viewing_duration").alias("max"),
                 sum("viewing_duration").alias("sum"),
                 avg("viewing_duration").alias("avg")) \
            .writeStream \
            .format("json") \
            .partitionBy("country") \
            .option("path", "./results/ex_5_3") \
            .option("checkpointLocation", checkpoint_path) \
            .start() \
            .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_6_1"
kafka_servers = "localhost:9092"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

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

adViewsDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe","adViews") \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schemaAdViews = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

adViewsStream = adViewsDF.select(from_json(col("value"), schemaAdViews).alias("t")) \
                .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
                .withWatermark("event_time", "10 seconds") 

adClicksDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe","adClicks") \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)")

schemaAdClicks = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True)
  ])

adClicksStream = adClicksDF.select(from_json(col("value"), schemaAdClicks).alias("t")) \
                .select(col("t.event_time").alias("click_time"), col("t.ad_id").alias("click_ad_id"), 
                        col("t.country").alias("click_contry"), col("t.site").alias("click_site")) \
                .withWatermark("click_time", "10 seconds")  

query = adViewsStream.join(adClicksStream, 
                    expr(""" 
                        click_ad_id = ad_id 
                        AND click_site = site
                        AND click_contry = country
                        AND click_time >= event_time
                        AND click_time <= event_time + interval 15 seconds    
                    """
                    )
                 ) \
                .select("event_time", "click_time", "ad_id", "site", "country", "viewing_duration") \
                .writeStream \
                .format("console") \
                .outputMode("append") \
                .option("checkpointLocation", checkpoint_path) \
                .start() \
                .awaitTermination()

import os

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'

checkpoint_path = "./checkpoint/ex_6_3"
kafka_servers = "localhost:9092"

import findspark
findspark.init('/home/andrey/spark-3.1.1-bin-hadoop3.2')

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

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

adViewsDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe","adViews") \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)") 

schemaAdViews = StructType([ \
    StructField("event_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True), \
    StructField("viewing_duration", IntegerType(), True) \
  ])

adViewsStream = adViewsDF.select(from_json(col("value"), schemaAdViews).alias("t")) \
                .select("t.event_time", "t.ad_id", "t.country", "t.site", "t.viewing_duration") \
                .withWatermark("event_time", "10 seconds") 

adClicksDF = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe","adClicks") \
            .option("startingOffsets", "latest") \
            .option("failOnDataLoss", "false") \
            .load() \
            .selectExpr("CAST(value AS STRING)")

schemaAdClicks = StructType([ \
    StructField("click_time", TimestampType(), True), \
    StructField("ad_id", IntegerType(), True), \
    StructField("country", StringType(), True), \
    StructField("site", StringType(), True)
  ])

adClicksStream = adClicksDF.select(from_json(col("value"), schemaAdClicks).alias("t")) \
                .select("t.click_time", col("t.ad_id").alias("click_ad_id"), 
                        col("t.country").alias("click_contry"), col("t.site").alias("click_site")) \
                .withWatermark("click_time", "20 seconds")  

query = adViewsStream.join(adClicksStream, 
                    expr(""" 
                        click_ad_id = ad_id 
                        AND click_site = site
                        AND click_contry = country
                        AND click_time >= event_time
                        AND click_time <= event_time + interval 15 seconds    
                    """
                    )
                 ) \
                .select(to_json(struct("event_time", "click_time", "ad_id", "site", "country", "viewing_duration")).alias("value")) \
                .writeStream \
                .format("kafka") \
                .option("kafka.bootstrap.servers", kafka_servers) \
                .option("topic","adStatistic") \
                .option("checkpointLocation", checkpoint_path) \
                .start() \
                .awaitTermination()