Τι συμβαίνει πραγματικά όταν εκτελείτε μια εργασία Spark;

Εισαγωγή

Καθώς τα σύνολα δεδομένων μεγαλώνουν, η αποτελεσματική επεξεργασία τους σε ένα μόνο μηχάνημα μπορεί να γίνει δύσκολη, λόγω των περιορισμών στην υπολογιστική ισχύ της CPU, στη μνήμη και στην ταχύτητα μεταφοράς δεδομένων από και προς τα μέσα αποθήκευσης.

Το Apache Spark αντιμετωπίζει αυτές τις προκλήσεις προσφέροντας μια μηχανή κατανεμημένης επεξεργασίας δεδομένων, η οποία μπορεί να χωρίζει τους φόρτους εργασίας σε διαμερίσματα και να εκτελεί εργασίες παράλληλα, σε πολλούς πυρήνες CPU και σε πολλά μηχανήματα.

Ωστόσο, το Spark δεν αφορά απλώς την προσθήκη περισσότερων μηχανημάτων.

Η κατανόηση του τρόπου με τον οποίο συνεργάζονται τα επιμέρους στοιχεία του —οδηγοί, εκτελεστές, RDD, DataFrames, διαμερίσματα, εργασίες, στάδια και επιμέρους διεργασίες— είναι απαραίτητη για τον σχεδιασμό αποδοτικών και αξιόπιστων διοχετεύσεων δεδομένων.

Σε αυτό το άρθρο, θα εξετάσω την αρχιτεκτονική του Spark, τους μηχανισμούς εκτέλεσης και βασικά ζητήματα απόδοσης, χρησιμοποιώντας ως πρακτικό παράδειγμα τη συνάθροιση δεδομένων πωλήσεων ηλεκτρονικού εμπορίου.

1. Γιατί το Apache Spark — Παράλληλη επεξεργασία δεδομένων

Το Apache Spark είναι μια μηχανή κατανεμημένων υπολογισμών, σχεδιασμένη για επεξεργασία δεδομένων μεγάλης κλίμακας και μηχανική δεδομένων.

Το Spark επεξεργάζεται τα μεγάλα δεδομένα αξιοποιώντας την παράλληλη επεξεργασία δεδομένων:

Σχήμα A. Αρχιτεκτονική συστάδας Apache Spark σε υψηλό επίπεδο (συστάδα, οδηγός και κόμβοι εργασίας), με σύγκριση της εκτέλεσης σε έναν μόνο κόμβο (δημιουργία: Kuriko IWAI)

Σε σύγκριση με την επεξεργασία σε έναν μόνο κόμβο (δεξιά, Σχήμα A), το Spark χωρίζει την εργασία σε επιμέρους διεργασίες, χρησιμοποιώντας πολλούς κόμβους και εκτελώντας τες στους διαθέσιμους πόρους πολλών μηχανημάτων.

Εδώ, ο όρος κατανεμημένο δεν σημαίνει ότι είναι υποχρεωτική η χρήση πολλών μηχανημάτων. Το Spark μπορεί να εκτελεστεί σε τοπική λειτουργία, σε ένα μόνο μηχάνημα. Ακόμη και σε αυτήν την περίπτωση, τα δεδομένα μπορούν να χωριστούν σε διαμερίσματα και να υποβληθούν σε παράλληλη επεξεργασία.

2. Βασική αρχιτεκτονική του Apache Spark

Το Spark αποτελείται από αρκετά βασικά στοιχεία:

• Οδηγός Spark (οδηγός): Η κεντρική διεργασία συντονισμού μιας εφαρμογής Spark.
• Διαχειριστής συστάδας: Εξωτερική υπηρεσία που είναι υπεύθυνη για την κατανομή πόρων υλικού, όπως πυρήνες CPU και μνήμη.
• Κόμβοι εργασίας (κόμβος εργασίας): Φυσικό ή εικονικό μηχάνημα ή υπολογιστικός κόμβος της συστάδας.
• Εκτελεστές: Κατανεμημένες διεργασίες εργασίας που εκκινούνται στους κόμβους εργασίας.
• Java Virtual Machine (JVM): Το περιβάλλον εκτέλεσης που απαιτείται για την εκτέλεση του bytecode του Spark (καθώς το Spark είναι γραμμένο σε Scala/Java).

Αρχικά, ο οδηγός συντονίζει μια εφαρμογή Spark και τη μετατρέπει σε κατευθυνόμενο ακυκλικό γράφο (DAG) φυσικών εργασιών εκτέλεσης, τις οποίες προγραμματίζει στους εκτελεστές.

Στη συνέχεια, ο διαχειριστής συστάδας κατανέμει πόρους στο πλαίσιο του συστήματος διαχείρισης πόρων της ανάπτυξης.

Οι κόμβοι εργασίας παρέχουν υπολογιστικούς πόρους στους εκτελεστές, οι οποίοι εκτελούν εργασίες και διατηρούν δεδομένα κατά την εκτέλεσή τους.

Ορισμένα διαχειριζόμενα περιβάλλοντα, όπως το Databricks, χρησιμοποιούν έναν εκτελεστή ανά κόμβο εργασίας, αλλά αυτό δεν αποτελεί καθολικό κανόνα του Spark. Ανάλογα με τη διαμόρφωση της συστάδας, ένας κόμβος εργασίας μπορεί να φιλοξενεί πολλούς εκτελεστές.

2-1. Περίπτωση χρήσης ηλεκτρονικού εμπορίου — Δομή δεδομένων εισόδου

Ας υποθέσουμε ότι υπάρχει μια μικρή εταιρεία ηλεκτρονικού εμπορίου.

Διαθέτει 10,000 εγγραφές πωλήσεων, καθεμία από τις οποίες περιλαμβάνει έναν κωδικό μονάδας αποθέματος (SKU) και την ποσότητα που πωλήθηκε. Ας υποθέσουμε ότι υπάρχουν τρία πιθανά SKU: A, B, και C.

Πίνακας 1. Παράδειγμα ανάλυσης εγγραφών πωλήσεων ηλεκτρονικού εμπορίου και αναμενόμενων συναθροίσεων.

Στόχος μας είναι να υπολογίσουμε τις συνολικές έγκυρες μονάδες που πωλήθηκαν ανά SKU:

1. Αποκλείστε τις εγγραφές των οποίων qty <= 0, στη συνέχεια
2. Συγκεντρώστε τα δείγματα ανά SKU.

Στο ενδεικτικό πλήρες σύνολο δεδομένων, ας υποθέσουμε ότι μετά το φιλτράρισμα απομένουν 9,800 από τις 10,000 εγγραφές. Τα τελικά σύνολα που αναμένονται από το Spark είναι:

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. Συστάδα Spark

Θα ορίσω μια απλοποιημένη συστάδα για την περίπτωσή μας.

Κόμβοι εργασίας

Θα ορίσω 3 κόμβους εργασίας, καθέναν με τύπο στιγμιότυπου 8 vCPU και μνήμη RAM 32 GiB:

Σχήμα B. Κατανομή πόρων στους κόμβους εργασίας (δημιουργία: Kuriko IWAI)

Συνολικά, στο σύστημα είναι διαθέσιμα 24 vCPU.

Κατανεμητές Spark και κατανομή CPU/μνήμης

Ας υποθέσουμε ότι εξασφαλίζω:

• 7 κατανεμητές.
• 2 πυρήνες vCPU για κάθε κατανεμητή.
• 3 GiB μνήμης σωρού JVM για κάθε κατανεμητή.

Η εφαρμογή Spark ζητά από τον διαχειριστή του συμπλέγματος να κατανείμει πόρους στους 7 αυτούς κατανεμητές.

Στη συνέχεια, ο διαχειριστής του συμπλέγματος αποφασίζει πώς θα κατανεμηθούν οι CPU και η μνήμη με βάση τη χρήση πόρων κάθε κόμβου εργασίας.

Για παράδειγμα, αποφασίζει να δημιουργήσει 14 θέσεις εργασιών — 2 πυρήνες vCPU ανά κατανεμητή x 7 κατανεμητές:

Εικόνα C-1. Πυρήνες vCPU, μνήμη και κατανομή κατανεμητών στους κόμβους εργασίας, με ανισορροπία πόρων στον κόμβο εργασίας A

Σημειώστε ότι τα 3 GiB μνήμης που κατανέμονται στον κατανεμητή δεν αποτελούν τη συνολική μνήμη του, επειδή πρέπει να εξασφαλιστεί επιπλέον μνήμη για τις επιβαρύνσεις των κόμβων εργασίας και άλλες εκτελέσεις.

Αν υποθέσουμε ότι μία εργασία απαιτεί 1 vCPU (spark.task.cpus = 1), κάθε κόμβος εργασίας κατανέμει τα vCPU του:

Εικόνα C-2. Παράλληλη κατανομή vCPU στους κόμβους εργασίας

Αυτό αναδεικνύει δύο βασικά σημεία:

1. Κάθε κόμβος εργασίας έχει τον δικό του ρυθμό χρήσης CPU (στην Εικόνα C-2, η χρήση του κόμβου εργασίας A είναι 75%, ενώ των κόμβων εργασίας B και C παραμένει στο 50%).
2. Παρόλο που το σύστημα διαθέτει 24 πυρήνες vCPU (8 vCPU/κόμβο εργασίας x 3 κόμβους εργασίας), μπορεί να χρησιμοποιήσει μόνο 14 πυρήνες vCPU· οι υπόλοιποι 10 παραμένουν αχρησιμοποίητοι για αυτές τις εργασίες.

2-3. Η Java Virtual Machine (JVM) πίσω από το Spark

Το Apache Spark βασίζεται στην Java Virtual Machine (JVM).

Η JVM είναι ένα περιβάλλον εκτέλεσης που απαιτείται για την εκτέλεση του bytecode του Spark.

Καθώς το Spark είναι γραμμένο σε Scala/Java:

• Όταν χρησιμοποιείτε Scala ή Java στην εφαρμογή: Ο κώδικας εκτελείται εγγενώς μέσα στη JVM.
• Όταν χρησιμοποιείτε PySpark (Python): Μια διεργασία Python εκτελεί τον κώδικα, αλλά επικοινωνεί με ένα πρόγραμμα οδήγησης βασισμένο σε JVM μέσω μιας γέφυρας όπως η Py4J. Οι κατανεμητές Spark στους κόμβους εργασίας εξακολουθούν να εκτελούνται μέσα σε JVM, η οποία στη συνέχεια διαχειρίζεται εργαζόμενες διεργασίες Python για την εκτέλεση κώδικα Python και συναρτήσεων που ορίζονται από τον χρήστη στην Python (Python UDFs).

Και στις δύο περιπτώσεις, η JVM διαχειρίζεται τα νήματα εκτέλεσης, την κατανομή μνήμης και την αποκομιδή απορριμμάτων τόσο για το πρόγραμμα οδήγησης όσο και για τους κατανεμητές.

JVM προγράμματος οδήγησης και κατανεμητών σε εφαρμογή PySpark

Σε μια εφαρμογή PySpark, το πρόγραμμα οδήγησης Python επικοινωνεί με το πρόγραμμα οδήγησης Spark που εκτελείται στη JVM·

• Πρόγραμμα οδήγησης Python: Εκεί όπου γράφεται και εκτελείται ο κώδικας PySpark σε επίπεδο εφαρμογής.
• JVM προγράμματος οδήγησης: Συντονίζει την εκτέλεση και τον προγραμματισμό εργασιών του Spark.
• JVM κατανεμητή: Εκτελεί εργασίες Spark και διαχειρίζεται τους πόρους εκτέλεσης. Όταν οι εργασίες απαιτούν εκτέλεση Python, συμμετέχουν εργαζόμενες διεργασίες Python.
• JVM πελάτη (διεργασία πελάτη ή μηχάνημα υποβολής): JVM που εκτελείται στον τοπικό υπολογιστή ή σε κόμβο ακμής/πύλης.

Όπως φαίνεται στο σχήμα, η JVM πελάτη συνδέεται πρώτα με την JVM του προγράμματος οδήγησης, η οποία στη συνέχεια συντονίζεται με τις JVM των κατανεμητών:

Εικόνα D. Συνολική σχέση μεταξύ των JVM πελάτη, προγράμματος οδήγησης και κατανεμητών

Από αυτό το σημείο και έπειτα, η JVM του προγράμματος οδήγησης φιλοξενεί το Spark Driver, ο οποίος διαχειρίζεται την εφαρμογή, ενώ οι JVM των κατανεμητών φιλοξενούν τους επιμέρους κατανεμητές που είναι υπεύθυνοι για την εκτέλεση αυτών των εργασιών.

Κάθε JVM προγράμματος οδήγησης διαθέτει σωρό JVM για την αποτελεσματική διαχείριση της επεξεργασίας δεδομένων, της προσωρινής αποθήκευσης και των αντικειμένων χρήστη.

2-4. Αναλυτική εξέταση των Resilient Distributed Datasets (RDD)

Ένα Resilient Distributed Dataset (RDD) είναι μια αμετάβλητη συλλογή στοιχείων χωρισμένη σε διαμερίσματα, την οποία το Spark μπορεί να επεξεργάζεται παράλληλα.

Τα RDD παρακολουθούν τις εξαρτήσεις μεταξύ των μετασχηματισμών, επιτρέποντας στο Spark να επανυπολογίζει τα διαμερίσματα που χάθηκαν σε πολλά σενάρια αστοχίας.

Η ονομασία περιγράφει τρεις ιδιότητες:

• Ανθεκτικό: Το Spark καταγράφει τη γραμμή προέλευσης —ποιο RDD εξαρτάται από ποιο προηγούμενο RDD— και συχνά μπορεί να ανακατασκευάσει τα διαμερίσματα που λείπουν, αντί να διατηρεί αντίγραφα κάθε ενδιάμεσου αποτελέσματος.
• Κατανεμημένο: Το σύνολο δεδομένων αναπαρίσταται ως διαμερίσματα, τα οποία το Spark μπορεί να επεξεργάζεται σε διαφορετικούς κατανεμητές. Δεν απαιτούνται πολλαπλά φυσικά μηχανήματα· η τοπική λειτουργία υποστηρίζει επίσης RDD.
• Σύνολο δεδομένων: Ένα RDD μπορεί να περιέχει ζεύγη κλειδιού-τιμής, ακέραιους αριθμούς, συμβολοσειρές, πλειάδες και άλλα υποστηριζόμενα στοιχεία — όχι μόνο σχεσιακές γραμμές. Τα API των RDD περιλαμβάνουν λειτουργίες όπως map, filter, reduceByKey, join, και aggregateByKey.

Ένα RDD είναι μια λογική αφαίρεση συλλογής και υπολογισμού, όχι μια δήλωση ότι συγκεκριμένα διαμερίσματα έχουν εκχωρηθεί μόνιμα σε συγκεκριμένους κόμβους εργασίας.

Διαμερίσματα

Ένα διαμέρισμα είναι ένα λογικό τμήμα δεδομένων και συνήθως αντιστοιχεί σε μία εργασία μέσα σε ένα στάδιο.

Ας υποθέσουμε ότι δημιουργούμε 100 διαμερίσματα (η επιλογή των 100 αρχικών διαμερισμάτων γίνεται ρητά για εκπαιδευτικούς σκοπούς).

Με 10.000 στοιχεία στην πηγή, αντιστοιχούν περίπου 100 στοιχεία ανά διαμέρισμα, εφόσον η είσοδος κατανέμεται σχετικά ομοιόμορφα:

Σχήμα E. Παράλληλη κατανομή διαμερισμάτων σε 100 αρχικά τμήματα δεδομένων

3. API RDD έναντι API DataFrame: εσωτερική βελτιστοποίηση Catalyst

Το αρχικό παράδειγμα με RDD βοηθά στην κατανόηση της γενεαλογίας, των διαμερισμάτων και του shuffle.

Ωστόσο, στις συνήθεις δομημένες διαδικασίες ETL, το API DataFrame είναι το προτιμώμενο σημείο εκκίνησης, επειδή παρέχει στήλες, πληροφορίες σχήματος, εκφράσεις τύπου SQL και λειτουργίες ορατές στον βελτιστοποιητή.

Το παρακάτω διάγραμμα παρουσιάζει τη σχέση:

Σχήμα F. Η σχέση μεταξύ των εννοιών γραφήματος του Spark: γενεαλογία RDD, DataFrame και DAG εκτέλεσης του Spark

Όπως το RDD έχει τη γενεαλογία RDD για να ορίζει τους μετασχηματισμούς, έτσι και το σχέδιο ερωτήματος του DataFrame αναπαριστά το δομημένο ερώτημά του.

Το Spark δημιουργεί τις εργασίες εκτέλεσης με βάση τις εξαρτήσεις και τους φυσικούς τελεστές.

Η εκτέλεση DataFrame μπορεί να χρησιμοποιεί εσωτερικά μηχανισμούς που βασίζονται σε RDD, αλλά αυτό δεν σημαίνει ότι οι χρήστες χρειάζεται να δημιουργούν χειροκίνητα τη γενεαλογία RDD όταν γράφουν κώδικα DataFrame.

3-1. Σχέδιο ερωτήματος DataFrame

Ένα σχέδιο ερωτήματος DataFrame είναι η ακολουθία λογικών και φυσικών βημάτων που δημιουργεί μια μηχανή εκτέλεσης (το μωβ πλαίσιο στο Σχήμα L).

Ένα ερώτημα DataFrame περνά από διάφορες αναπαραστάσεις:

1. Λογικό σχέδιο: Περιγραφή των ζητούμενων λειτουργιών (σάρωση, φιλτράρισμα, ομαδοποίηση, συνάθροιση).
2. Αναλυμένο και βελτιστοποιημένο λογικό σχέδιο: Το Spark επιλύει τις αναφορές και εφαρμόζει τις κατάλληλες λογικές βελτιστοποιήσεις, όπως την αποκοπή στηλών ή την προώθηση συνθηκών φιλτραρίσματος.
3. Φυσικό σχέδιο: Το Spark επιλέγει εκτελέσιμους τελεστές, π.χ. σαρώσεις, μερικές/τελικές συναθροίσεις, ανταλλαγές, συνδέσεις και ταξινομήσεις.
4. Εκτέλεση: Το φυσικό σχέδιο εκτελείται με τους μηχανισμούς εκτέλεσης του Spark και, τελικά, περιλαμβάνει εργασίες σε εκτελεστές.

Στην περίπτωσή μας, το φυσικό σχέδιο μπορεί να περιλαμβάνει τα ακόλουθα βήματα:

• Μερική συνάθροιση κατακερματισμού,
• Ανταλλαγή με βάση το SKU και
• Τελική συνάθροιση κατακερματισμού.

3-2. Σύγκριση ενεργειών: API RDD έναντι API DataFrame

Ας συγκρίνουμε τα API RDD και DataFrame, χρησιμοποιώντας ως παράδειγμα τη συνάθροιση ανά SKU.

API RDD

Τα στοιχεία είναι (sku, qty) ζεύγη. Η λογική συνάθροισης σε Python είναι ρητή και το Spark ακολουθεί τις εξαρτήσεις των RDD:

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

API DataFrame

Ας υποθέσουμε ότι sales_df έχει στήλες sku (συμβολοσειρά) και qty (αριθμητική):

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

Εδώ, και τα δύο παραδείγματα επιλύουν το ίδιο επιχειρηματικό πρόβλημα:

Φιλτράρισμα μη έγκυρων εγγραφών πωλήσεων και συνάθροιση των έγκυρων ποσοτήτων ανά SKU.

Και στις δύο περιπτώσεις, collect() ενεργοποιεί την εκτέλεση του Spark και επιστρέφει όλες τις εγγραφές που προκύπτουν στο Driver.

Η βασική διαφορά εδώ αφορά την αναπαράσταση των αποτελεσμάτων:

• API RDD: Επιστρέφει στοιχεία Python, όπως (sku, total_qty) πλειάδες.
• API DataFrame: Επιστρέφει δομημένα Row αντικείμενα με ονομασμένα πεδία, όπως sku και total_qty.

Ακολουθεί μια συνολική σύγκριση των API RDD και DataFrame:

Πίνακας 2. RDD API έναντι DataFrame API — σύγκριση χαρακτηριστικών και επιπέδου αφαίρεσης: (Δημιουργήθηκε από την Kuriko IWAI)

4. Συμπερασματικά

Στόχος της ρύθμισης του Spark δεν είναι να μεγιστοποιήσει ανεξάρτητα τον αριθμό των διαμερισμάτων ή τη χρήση της CPU.

Στόχος είναι να μειώσει το κόστος εκτέλεσης από άκρο σε άκρο και την καθυστέρηση, χωρίς να προκαλέσει αστάθεια.

Σε αυτό το άρθρο, μαθαίνουμε τις βασικές σχέσεις στο σύστημα του Spark:

• Πόροι: Οι κόμβοι εργασίας φιλοξενούν εκτελεστές· οι εκτελεστές εκτελούν επιμέρους εργασίες· οι ρυθμισμένοι πόροι CPU και μνήμης περιορίζουν τον βαθμό παραλληλισμού.
• Προγραμματισμός με RDD: Οι μετασχηματισμοί παράγουν αμετάβλητους ορισμούς RDD και τη γενεαλογία τους· οι ενέργειες ενεργοποιούν τον υπολογισμό.
• Προγραμματισμός με DataFrame: Οι μετασχηματισμοί παράγουν λογικές λειτουργίες ερωτημάτων· η βελτιστοποίηση επιλέγει εκτελέσιμους φυσικούς τελεστές· οι ενέργειες ενεργοποιούν τον υπολογισμό.
• Εκτέλεση: Το Spark οργανώνει την εργασία σε εργασίες, στάδια και επιμέρους εργασίες, ενώ οι επιμέρους εργασίες επεξεργάζονται διαμερίσματα στους εκτελεστές. Οι ανακατανομές (shuffles) και άλλες εξαρτήσεις δημιουργούν όρια συντονισμού μεταξύ των σταδίων.

Η κατανόηση του τρόπου με τον οποίο το Spark εκτελεί κατανεμημένους φόρτους εργασίας αποτελεί μόνο τη βάση.

Μπορείτε να μάθετε μια ροή υπολογισμού επτά βημάτων με χρήση του RDD API.