אפאצ'י קפקא ועיבוד נתונים בסטרימינג עם Spark Streaming

שלום, הבר! היום נבנה מערכת שתעבד את זרמי ההודעות של Apache Kafka באמצעות Spark Streaming ותכתוב את תוצאות העיבוד למסד הנתונים בענן AWS RDS.

בואו נדמיין שמוסד אשראי מסוים מטיל עלינו את המשימה של עיבוד עסקאות נכנסות "בתנועה" בכל סניפיו. ניתן לעשות זאת לצורך חישוב מיידי של פוזיציית מטבע פתוחה לאוצר, מגבלות או תוצאות פיננסיות לעסקאות וכו'.

איך ליישם את המקרה הזה ללא שימוש בכישופי קסמים וכשפים - קרא מתחת לגזירה! ללכת!

(מקור תמונה)

מבוא

כמובן, עיבוד כמות גדולה של נתונים בזמן אמת מספק הזדמנויות רבות לשימוש במערכות מודרניות. אחד השילובים הפופולריים ביותר לכך הוא הטנדם של Apache Kafka ו-Spark Streaming, שבו קפקא יוצר זרם של חבילות הודעות נכנסות, ו-Spark Streaming מעבד מנות אלו במרווח זמן נתון.

כדי להגביר את סבילות התקלות של האפליקציה, נשתמש בנקודות ביקורת. עם המנגנון הזה, כאשר מנוע ה-Spark Streaming צריך לשחזר נתונים שאבדו, הוא רק צריך לחזור לנקודת המחסום האחרונה ולחדש את החישובים משם.

ארכיטקטורה של המערכת שפותחה

רכיבים בשימוש:

• אפאצ'י קפקא היא מערכת מבוזרת לפרסום-הרשמה להודעות. מתאים גם לצריכת הודעות במצב לא מקוון וגם לצריכת הודעות מקוונת. כדי למנוע אובדן נתונים, הודעות קפקא מאוחסנות בדיסק ומשוכפלות בתוך האשכול. מערכת קפקא בנויה על גבי שירות הסנכרון ZooKeeper;
• הזרמת אפאצ'י ספארק - רכיב Spark לעיבוד נתונים זורמים. מודול Spark Streaming בנוי באמצעות ארכיטקטורת מיקרו-אצווה, שבה זרם הנתונים מתפרש כרצף רציף של מנות נתונים קטנות. Spark Streaming לוקח נתונים ממקורות שונים ומשלב אותם בחבילות קטנות. חבילות חדשות נוצרות במרווחי זמן קבועים. בתחילת כל מרווח זמן, נוצרת חבילה חדשה, וכל הנתונים המתקבלים במהלך אותו מרווח זמן נכללים בחבילה. בתום המרווח, צמיחת החבילה נעצרת. גודל המרווח נקבע על ידי פרמטר הנקרא מרווח אצווה;
• Apache Spark SQL - משלב עיבוד יחסי עם תכנות פונקציונלי של Spark. נתונים מובנים פירושם נתונים שיש להם סכימה, כלומר קבוצה אחת של שדות עבור כל הרשומות. Spark SQL תומך בקלט ממגוון מקורות נתונים מובנים, ובזכות הזמינות של מידע סכימה, הוא יכול לאחזר ביעילות רק את שדות הרשומות הנדרשים, וכן מספק ממשקי DataFrame API;
• AWS RDS הוא מסד נתונים יחסי מבוסס ענן זול יחסית, שירות אינטרנט המפשט הגדרה, תפעול וקנה מידה, ומנוהל ישירות על ידי אמזון.

התקנה והרצה של שרת קפקא

לפני השימוש ישירות בקפקא, עליך לוודא שיש לך 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

לפני הפעלת שרת Kafka, עליך להפעיל את שרת ZooKeeper; אנו נשתמש בסקריפט העזר שמגיע עם הפצת Kafka:

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

לאחר הפעלת ZooKeeper בהצלחה, הפעל את שרת Kafka במסוף נפרד:

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 באמצעות ה-API של KafkaProducer.

כותב מפיק

המפיק יפיק נתונים אקראיים - 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

אפאצ 'י ספארק היא פלטפורמת מחשוב אשכולות אוניברסלית ובעלת ביצועים גבוהים.

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

לאחר מכן, שמים סימון בגוש Free Tier, ולאחר מכן יוצע לנו אוטומטית מופע של מחלקת t2.micro - למרות שהיא חלשה, היא חינמית ודי מתאימה למשימה שלנו:

בהמשך מגיעים דברים חשובים מאוד: שם מופע בסיס הנתונים, שם המשתמש הראשי והסיסמה שלו. בואו נקרא למופע: myHabrTest, משתמש ראשי: האבר, סיסמה: habr12345 ולחץ על כפתור הבא:

בעמוד הבא יש פרמטרים האחראים על הנגישות של שרת מסד הנתונים שלנו מבחוץ (Public accessibility) וזמינות הפורטים:

בואו ניצור הגדרה חדשה לקבוצת האבטחה VPC, שתאפשר גישה חיצונית לשרת מסד הנתונים שלנו דרך יציאה 5432 (PostgreSQL).
בואו נעבור למסוף AWS בחלון דפדפן נפרד ללוח המחוונים של VPC -> קבוצות אבטחה -> יצירת קבוצת אבטחה:

אנו מגדירים את השם לקבוצת האבטחה - PostgreSQL, תיאור, מציינים לאיזה VPC קבוצה זו צריכה להיות משויכת ולחץ על כפתור צור:

מלא את הכללים הנכנסים עבור יציאה 5432 עבור הקבוצה החדשה שנוצרה, כפי שמוצג בתמונה למטה. אתה לא יכול לציין את היציאה באופן ידני, אבל בחר PostgreSQL מהרשימה הנפתחת סוג.

באופן קפדני, הערך ::/0 פירושו זמינות של תעבורה נכנסת לשרת מכל רחבי העולם, מה שקנוני לא לגמרי נכון, אבל כדי לנתח את הדוגמה, בואו נאפשר לעצמנו להשתמש בגישה הזו:

אנו חוזרים לדף הדפדפן, שם נפתח "הגדר הגדרות מתקדמות" ובחר בסעיף קבוצות האבטחה של VPC -> בחר קבוצות אבטחה קיימות של VPC -> PostgreSQL:

לאחר מכן, באפשרויות מסד הנתונים -> שם מסד הנתונים -> הגדר את השם - habrDB.

אנו יכולים להשאיר את שאר הפרמטרים, למעט ביטול גיבוי (תקופת שמירת גיבוי - 0 ימים), ניטור ותובנות ביצועים, כברירת מחדל. לחץ על הכפתור צור בסיס נתונים:

מטפל בחוטים

השלב האחרון יהיה פיתוח עבודה של Spark, שתעבד נתונים חדשים המגיעים מקפקא כל שתי שניות ותכניס את התוצאה למסד הנתונים.

כפי שצוין לעיל, מחסומים הם מנגנון ליבה ב-SparkStreaming שיש להגדיר אותו כדי להבטיח סובלנות לתקלות. אנו נשתמש בנקודות ביקורת, ואם ההליך נכשל, מודול Spark Streaming יצטרך רק לחזור לנקודת הבידוק האחרונה ולחדש ממנו חישובים כדי לשחזר את הנתונים שאבדו.

ניתן להפעיל את המחסום על ידי הגדרת ספרייה במערכת קבצים עמידה לתקלות ואמינה (כגון HDFS, S3 וכו') שבה יישמר פרטי המחסום. זה נעשה באמצעות, למשל:

streamingContext.checkpoint(checkpointDirectory)

בדוגמה שלנו, נשתמש בגישה הבאה, כלומר, אם checkpointDirectory קיים, אז ההקשר ייווצר מחדש מנתוני המחסום. אם הספרייה לא קיימת (כלומר מבוצעת בפעם הראשונה), אזי נקרא functionToCreateContext כדי ליצור הקשר חדש ולהגדיר את DStreams:

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 ככתובת האתר של שרת מסד הנתונים, המוצגת בסעיף קישוריות ואבטחה:

על מנת לחבר נכון את ספארק וקפקא, עליך להריץ את העבודה באמצעות smark-submit באמצעות החפץ spark-streaming-kafka-0-8_2.11. בנוסף, נשתמש גם בחפץ לאינטראקציה עם מסד הנתונים PostgreSQL; נעביר אותם באמצעות --packages.

לצורך גמישות הסקריפט נכלול כפרמטרי קלט גם את שם שרת ההודעות והנושא ממנו נרצה לקבל נתונים.

אז, הגיע הזמן להפעיל ולבדוק את הפונקציונליות של המערכת:

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