Апачи Кафка и стриминг обработка на податоци со стриминг на искра

Здраво, Хабр! Денеска ќе изградиме систем кој ќе ги обработува преносите на пораките на Apache Kafka со помош на Spark Streaming и ќе ги запише резултатите од обработката во базата на податоци на облакот AWS RDS.

Да замислиме дека одредена кредитна институција ни поставува задача да ги обработуваме дојдовните трансакции „во лет“ низ сите нејзини филијали. Ова може да се направи со цел навремено пресметување на отворена девизна позиција за трезорот, лимити или финансиски резултати за трансакции итн.

Како да го имплементирате овој случај без употреба на магија и магични магии - прочитајте под рез! Оди!

(Извор на слика)

Вовед

Се разбира, обработката на голема количина на податоци во реално време дава многу можности за употреба во современите системи. Една од најпопуларните комбинации за ова е тандемот на Apache Kafka и Spark Streaming, каде што Кафка создава проток на пакети за дојдовни пораки, а Spark Streaming ги обработува овие пакети во даден временски интервал.

За да ја зголемиме толеранцијата на грешка на апликацијата, ќе користиме контролни точки. Со овој механизам, кога моторот Spark Streaming треба да ги врати изгубените податоци, треба само да се врати на последната контролна точка и да продолжи со пресметките од таму.

Архитектура на развиениот систем

Користени компоненти:

• Апачи Кафка е дистрибуиран систем за пораки за објавување-претплати. Погоден и за офлајн и за онлајн потрошувачка на пораки. За да се спречи губење на податоци, пораките на Кафка се зачувуваат на дискот и се реплицираат во кластерот. Системот Кафка е изграден на врвот на услугата за синхронизација ZooKeeper;
• Стриминг на Apache Spark - Spark компонента за обработка на стриминг податоци. Модулот Spark Streaming е изграден со микро-бач архитектура, каде што протокот на податоци се толкува како континуирана низа од мали пакети со податоци. Spark Streaming зема податоци од различни извори и ги комбинира во мали пакети. Новите пакети се креираат во редовни интервали. На почетокот на секој временски интервал се креира нов пакет и сите податоци добиени за време на тој интервал се вклучени во пакетот. На крајот на интервалот, растот на пакетите престанува. Големината на интервалот се одредува со параметар наречен сериски интервал;
• Apache Spark SQL - комбинира релациона обработка со функционално програмирање Spark. Структурни податоци значи податоци што имаат шема, односно единствен сет на полиња за сите записи. Spark SQL поддржува внесување од различни структурирани извори на податоци и, благодарение на достапноста на информации за шемата, може ефикасно да ги врати само потребните полиња на записи, а исто така обезбедува DataFrame API;
• AWS RDS е релативно евтина релациона база на податоци заснована на облак, веб-услуга што го поедноставува поставувањето, работењето и скалирањето, а се администрира директно од Amazon.

Инсталирање и водење на серверот Кафка

Пред директно да го користите Кафка, треба да бидете сигурни дека имате Java, бидејќи ... JVM се користи за работа:

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

Ајде да создадеме нов корисник за работа со Кафка:

sudo useradd kafka -m
sudo passwd kafka
sudo adduser kafka sudo

Следно, преземете ја дистрибуцијата од официјалната веб-страница на Apache Kafka:

wget -P /YOUR_PATH "http://apache-mirror.rbc.ru/pub/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz"

Отпакувајте ја преземената архива:

tar -xvzf /YOUR_PATH/kafka_2.12-2.2.0.tgz
ln -s /YOUR_PATH/kafka_2.12-2.2.0 kafka

Следниот чекор е опционален. Факт е дека стандардните поставки не ви дозволуваат целосно да ги користите сите можности на Apache Kafka. На пример, избришете тема, категорија, група на која може да се објавуваат пораки. За да го промениме ова, ајде да ја уредиме конфигурациската датотека:

vim ~/kafka/config/server.properties

Додадете го следново на крајот од датотеката:

delete.topic.enable = true

Пред да го стартувате серверот Кафка, треба да го стартувате серверот ZooKeeper; ние ќе ја користиме помошната скрипта што доаѓа со дистрибуцијата на Кафка:

Cd ~/kafka
bin/zookeeper-server-start.sh config/zookeeper.properties

Откако ZooKeeper ќе започне успешно, стартувајте го серверот Кафка во посебен терминал:

bin/kafka-server-start.sh config/server.properties

Ајде да создадеме нова тема наречена Трансакција:

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic transaction

Ајде да се увериме дека е создадена тема со потребниот број на партиции и репликација:

bin/kafka-topics.sh --describe --zookeeper localhost:2181

Да ги пропуштиме моментите на тестирање на производителот и потрошувачот за новосоздадената тема. Повеќе детали за тоа како можете да тестирате испраќање и примање пораки се напишани во официјалната документација - Испратете неколку пораки. Па, продолжуваме да пишуваме продуцент во Python користејќи го KafkaProducer API.

Пишување на продуцентот

Производителот ќе генерира случајни податоци - 100 пораки секоја секунда. Под случајни податоци подразбираме речник кој се состои од три полиња:

• Гранка - име на продажното место на кредитната институција;
• Валута — валута на трансакцијата;
• износ — износот на трансакцијата. Износот ќе биде позитивен број доколку се работи за купување на валута од страна на Банката, а негативен број доколку се работи за продажба.

Кодот за производителот изгледа вака:

from numpy.random import choice, randint

def get_random_value():
    new_dict = {}

    branch_list = ['Kazan', 'SPB', 'Novosibirsk', 'Surgut']
    currency_list = ['RUB', 'USD', 'EUR', 'GBP']

    new_dict['branch'] = choice(branch_list)
    new_dict['currency'] = choice(currency_list)
    new_dict['amount'] = randint(-100, 100)

    return new_dict

Следно, користејќи го методот за испраќање, испраќаме порака до серверот, до темата што ни треба, во JSON формат:

from kafka import KafkaProducer    

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

try:
    future = producer.send(topic = my_topic, value = data)
    record_metadata = future.get(timeout=10)
    
    print('--> The message has been sent to a topic: 
            {}, partition: {}, offset: {}' 
            .format(record_metadata.topic,
                record_metadata.partition,
                record_metadata.offset ))   
                             
except Exception as e:
    print('--> It seems an Error occurred: {}'.format(e))

finally:
    producer.flush()

Кога ја извршуваме скриптата, ги добиваме следните пораки во терминалот:

Ова значи дека сè работи како што сакавме - продуцентот генерира и испраќа пораки до темата што ни треба.
Следниот чекор е да го инсталирате Spark и да го обработите овој пренос на пораки.

Инсталирање на Apache Spark

Apache Spark е универзална и со високи перформанси кластер компјутерска платформа.

Spark работи подобро од популарните имплементации на моделот MapReduce додека поддржува поширок опсег на типови на пресметки, вклучувајќи интерактивни барања и обработка на стримови. Брзината игра важна улога при обработка на големи количини на податоци, бидејќи брзината е таа што ви овозможува да работите интерактивно без да трошите минути или часови на чекање. Една од најголемите предности на Spark што го прави толку брз е неговата способност да врши пресметки во меморијата.

Оваа рамка е напишана во Scala, така што прво треба да ја инсталирате:

sudo apt-get install scala

Преземете ја дистрибуцијата на Spark од официјалната веб-страница:

wget "http://mirror.linux-ia64.org/apache/spark/spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz"

Отпакувајте ја архивата:

sudo tar xvf spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz -C /usr/local/spark

Додадете ја патеката до Spark во датотеката bash:

vim ~/.bashrc

Додадете ги следните редови преку уредникот:

SPARK_HOME=/usr/local/spark
export PATH=$SPARK_HOME/bin:$PATH

Извршете ја командата подолу откако ќе направите промени во bashrc:

source ~/.bashrc

Распоредување на AWS PostgreSQL

Останува само да ја распоредиме базата на податоци во која ќе ги прикачуваме обработените информации од стримовите. За ова ќе ја користиме услугата AWS RDS.

Одете во конзолата AWS -> AWS RDS -> Бази на податоци -> Креирај база на податоци:

Изберете PostgreSQL и кликнете Следно:

Бидејќи Овој пример е само за едукативни цели; ние ќе користиме бесплатен сервер „на минимум“ (бесплатно ниво):

Следно, ставаме штиклирање во блокот Free Tier, а потоа автоматски ќе ни биде понуден примерок од класата t2.micro - иако слаб, тој е бесплатен и сосема погоден за нашата задача:

Следуваат многу важни работи: името на примерот на базата на податоци, името на главниот корисник и неговата лозинка. Ајде да го именуваме примерот: myHabrTest, главен корисник: хабр, лозинка: habr12345 и кликнете на копчето Следно:

На следната страница има параметри одговорни за пристапноста на нашиот сервер за база на податоци однадвор (Јавна пристапност) и достапноста на портата:

Ајде да создадеме нова поставка за безбедносната група VPC, која ќе овозможи надворешен пристап до нашиот сервер за бази на податоци преку портата 5432 (PostgreSQL).
Ајде да одиме во конзолата AWS во посебен прозорец на прелистувачот до контролната табла VPC -> Безбедносни групи -> Креирај секција за безбедносна група:

Го поставивме името за групата за безбедност - PostgreSQL, опис, наведете со кој VPC треба да се поврзе оваа група и кликнете на копчето Креирај:

Пополнете ги правилата за влез за портата 5432 за новосоздадената група, како што е прикажано на сликата подолу. Не можете рачно да ја одредите портата, но изберете PostgreSQL од паѓачката листа Тип.

Строго кажано, вредноста ::/0 значи достапност на дојдовен сообраќај до серверот од целиот свет, што канонски не е целосно точно, но за да го анализираме примерот, да си дозволиме да го користиме овој пристап:

Се враќаме на страницата на прелистувачот, каде што имаме отворено „Конфигурирај напредни поставки“ и избираме во делот VPC безбедносни групи -> Изберете постоечки VPC безбедносни групи -> PostgreSQL:

Следно, во опциите на базата на податоци -> Име на базата на податоци -> поставете го името - habrDB.

Можеме да ги оставиме преостанатите параметри, со исклучок на оневозможување на резервната копија (период на задржување на резервната копија - 0 дена), следење и Увид во перформансите, стандардно. Кликнете на копчето Креирај база на податоци:

Ракувач со конци

Последната фаза ќе биде развој на работа Spark, која ќе обработува нови податоци кои доаѓаат од Кафка на секои две секунди и ќе ги внесе резултатите во базата на податоци.

Како што е наведено погоре, контролните точки се основен механизам во SparkStreaming што мора да се конфигурира за да се обезбеди толеранција на грешки. Ќе користиме контролни точки и, ако процедурата не успее, модулот Spark Streaming ќе треба само да се врати на последната контролна точка и да продолжи со пресметките од него за да ги врати изгубените податоци.

Контролната точка може да се овозможи со поставување директориум на доверлив датотечен систем толерантен за грешки (како HDFS, S3, итн.) во кој ќе се складираат информациите за контролната точка. Ова се прави со користење, на пример:

streamingContext.checkpoint(checkpointDirectory)

Во нашиот пример, ќе го користиме следниов пристап, имено, ако постои Директориум за контролни точки, тогаш контекстот ќе се рекреира од податоците за контролната точка. Ако директориумот не постои (т.е. се извршува за прв пат), тогаш функцијатаToCreateContext се повикува да создаде нов контекст и да го конфигурира DSstreams:

from pyspark.streaming import StreamingContext

context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)

Ние создаваме објект DirectStream за да се поврземе со темата „трансакција“ користејќи го методот createDirectStream од библиотеката KafkaUtils:

from pyspark.streaming.kafka import KafkaUtils
    
sc = SparkContext(conf=conf)
ssc = StreamingContext(sc, 2)

broker_list = 'localhost:9092'
topic = 'transaction'

directKafkaStream = KafkaUtils.createDirectStream(ssc,
                                [topic],
                                {"metadata.broker.list": broker_list})

Парсирање на дојдовните податоци во JSON формат:

rowRdd = rdd.map(lambda w: Row(branch=w['branch'],
                                       currency=w['currency'],
                                       amount=w['amount']))
                                       
testDataFrame = spark.createDataFrame(rowRdd)
testDataFrame.createOrReplaceTempView("treasury_stream")

Користејќи Spark SQL, правиме едноставно групирање и го прикажуваме резултатот во конзолата:

select 
    from_unixtime(unix_timestamp()) as curr_time,
    t.branch                        as branch_name,
    t.currency                      as currency_code,
    sum(amount)                     as batch_value
from treasury_stream t
group by
    t.branch,
    t.currency

Добивање на текстот за барање и негово извршување преку Spark SQL:

sql_query = get_sql_query()
testResultDataFrame = spark.sql(sql_query)
testResultDataFrame.show(n=5)

И тогаш ги зачувуваме добиените збирни податоци во табела во AWS RDS. За да ги зачуваме резултатите од агрегацијата во табела со база на податоци, ќе го користиме методот за запишување на објектот DataFrame:

testResultDataFrame.write 
    .format("jdbc") 
    .mode("append") 
    .option("driver", 'org.postgresql.Driver') 
    .option("url","jdbc:postgresql://myhabrtest.ciny8bykwxeg.us-east-1.rds.amazonaws.com:5432/habrDB") 
    .option("dbtable", "transaction_flow") 
    .option("user", "habr") 
    .option("password", "habr12345") 
    .save()

Неколку зборови за поставување врска со AWS RDS. Ги создадовме корисникот и лозинката за него на чекорот „Имплементирање на AWS PostgreSQL“. Треба да користите Endpoint како URL-адреса на серверот на базата на податоци, што е прикажано во делот Поврзување и безбедност:

За правилно поврзување на Spark и Kafka, треба да ја извршите работата преку smark-submit користејќи го артефактот искра-стриминг-кафка-0-8_2.11. Дополнително, ќе користиме и артефакт за интеракција со базата на податоци PostgreSQL; ќе ги пренесеме преку --пакети.

За флексибилност на скриптата, како влезни параметри ќе го вклучиме и името на серверот за пораки и темата од која сакаме да примаме податоци.

Значи, време е да се стартува и да се провери функционалноста на системот:

spark-submit 
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2,
org.postgresql:postgresql:9.4.1207 
spark_job.py localhost:9092 transaction

Сè успеа! Како што можете да видите на сликата подолу, додека апликацијата работи, на секои 2 секунди се појавуваат нови резултати од агрегација, бидејќи го поставивме интервалот на серии на 2 секунди кога го креиравме објектот StreamingContext:

Следно, правиме едноставно барање до базата на податоци за да го провериме присуството на записи во табелата трансакција_проток:

Заклучок

Оваа статија разгледа пример за обработка на информации со помош на Spark Streaming во врска со Apache Kafka и PostgreSQL. Со растот на податоците од различни извори, тешко е да се прецени практичната вредност на Spark Streaming за креирање стриминг и апликации во реално време.

Целосниот изворен код можете да го најдете во моето складиште на GitHub.

Среќен сум што разговарам за оваа статија, со нетрпение ги очекувам вашите коментари, а исто така се надевам на конструктивна критика од сите грижливи читатели.

Ви посакувам успех!

П.с. Првично беше планирано да се користи локална база на податоци PostgreSQL, но со оглед на мојата љубов кон AWS, решив да ја преместам базата на податоци во облакот. Во следната статија на оваа тема, ќе покажам како да го имплементирам целиот систем опишан погоре во AWS користејќи AWS Kinesis и AWS EMR. Следете ги вестите!

Извор: www.habr.com