SPARK

Învață Apache Spark & PySpark: comenzi explicate cu exemple

70 de comenzi 16 categorii Comunitate & Q&A

Apache Spark este cel mai avansat și utilizat motor unificat de calcul distribuit pentru procesarea pe scară largă a datelor (Big Data) și învățare automată. Înlocuind definitiv vechiul model MapReduce din Hadoop (care scria fiecare pas intermediar lent pe disc), Spark execută calculele direct în memorie (In-Memory Processing), fiind de până la 100 de ori mai rapid.

Prin intermediul API-ului său cel mai popular, PySpark (legătura nativă Python), echipele de inginerie a datelor și oamenii de știință a datelor (Data Scientists) pot manipula cu ușurință petabytes de informație folosind structuri de date de nivel înalt (DataFrames), interogări Spark SQL optimizate automat de motorul Catalyst Optimizer și procesare continuă cu Structured Streaming pe arhitecturi moderne Lakehouse (Delta Lake).

Arhitectura Distribuită Master-Worker în Apache Spark

# Driver Program (Punctul de Intrare / SparkSession)
# ┌─────────────────────────────────────────────────────────────┐
# │ SparkSession ──► Catalyst Optimizer ──► DAG Scheduler      │
# └────────────────────────────┬────────────────────────────────┘
#                              │ (Instrucțiuni de Task pe Rețea)
#                              ▼
# ┌─────────────────────────────────────────────────────────────┐
# │ Cluster Manager (Kubernetes, YARN, Spark Standalone)        │
# └──────┬──────────────────────┬──────────────────────┬────────┘
#        │                      │                      │
#        ▼                      ▼                      ▼
# ┌──────────────┐     ┌──────────────┐     ┌──────────────┐
# │ Worker Node  │     │ Worker Node  │     │ Worker Node  │
# │ ┌──────────┐ │     │ ┌──────────┐ │     │ ┌──────────┐ │
# │ │Executor  │ │     │ │Executor  │ │     │ │Executor  │ │
# │ │Task Task │ │     │ │Task Task │ │     │ │Task Task │ │
# │ │In-Memory │ │     │ │In-Memory │ │     │ │In-Memory │ │
# │ └──────────┘ │     │ └──────────┘ │     │ └──────────┘ │
# └──────────────┘     └──────────────┘     └──────────────┘

Pilonii Tehnologici ai Apache Spark

Componentă Spark	Mecanism Tehnic	Rol în Ingineria Datelor
Spark DataFrames	Colecții distribuite de date organizate în coloane denumite cu scheme stricte.	Oferă o sintaxă expresivă asemănătoare cu Pandas, dar scalabilă pe mii de noduri de calcul.
Catalyst Optimizer	Motor de optimizare a planurilor de execuție prin rescrierea inteligentă a interogărilor.	Aplică automat predicate pushdown, proiecții de coloane și optimizări de tip broadcast join.
Tungsten Engine	Management manual al memoriei off-heap în format binar și generare de cod la cald (Whole-Stage CodeGen).	Elimină pauzele de Garbage Collection pe JVM și optimizează calculele direct în cache-ul procesorului L1/L2.
Structured Streaming	Procesare de fluxuri în timp real modelate ca tabele care cresc continuu.	Garanții stricte de procesare Exactly-Once cu integrare nativă pentru Apache Kafka și Delta Lake.

Script PySpark: Citire Parquet, Filtrare și Agregare Distribuită

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, sum, avg

# 1. Inițializarea sesiunii Spark:
spark = SparkSession.builder \
    .appName("DexblogEnterpriseAnalytics") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# 2. Citirea unui set masiv de date Parquet direct din cloud (S3 / HDFS):
df = spark.read.parquet("s3a://data-lake/transactions/year=2026/")

# 3. Transformări declarative optimizate de Catalyst:
summary = df.filter(col("status") == "COMPLETED") \
    .groupBy("merchant_category") \
    .agg(
        count("transaction_id").alias("total_count"),
        sum("amount").alias("total_volume"),
        avg("amount").alias("avg_ticket")
    ) \
    .orderBy(col("total_volume").desc())

# 4. Afișarea primelor 10 rezultate agregate:
summary.show(10, truncate=False)

spark.stop()

Arhitectură & Catalyst 1 comandă

Arhitectura Internă Apache Spark: De la RDD-uri la DataFrames, Planuri Logice și Optimizatorul Catalyst df.explain(extended=True) Ghid tehnic pentru arhitectura Spark: cum compilează Catalyst planul logic nesoluționat în plan fizic optimizat, generarea de cod pe întreg stadiul (Tungsten) și de ce DataFrames bat RDD-urile.

io 8 comenzi

PySpark & Operațiuni 1 comandă

Procesare de Date cu PySpark: Transformări (map, filter, join), Acțiuni (collect, count) și Evaluare Leneșă (Lazy Evaluation) df.filter(col('age') > 25).groupBy('country').count() Ghid complet pentru programarea în PySpark: diferențele între transformări înguste (narrow) și largi (wide transformations ce declanșează shuffle), evaluarea leneșă și acțiuni de execuție.

transformations 7 comenzi

Performanță & Tuning 1 comandă

Tuning și Performanță în Spark SQL: Partitioning, Bucketing, Broadcast Joins și Evitarea Shuffling-ului Costisitor from pyspark.sql.functions import broadcast; large_df.join(broadcast(small_df), 'key') Ghid de optimizare a performanței în Spark: de ce operațiunea de Shuffle este cel mai mare inamic al vitezei, utilizarea broadcast(small_df) pentru uniri rapide în memorie, Adaptive Query Execution (AQE) și particionare.

Structured Streaming & Delta Lake 1 comandă

Ingestie și Streaming de Date cu Spark Structured Streaming: Integrare cu Apache Kafka și Arhitectura Medallion pe Delta Lake df.writeStream.format('delta').outputMode('append').start('/lakehouse/bronze') Ghid enterprise pentru arhitecturi moderne Lakehouse: citirea fluxurilor de date din Apache Kafka, scrierea tranzacțională în format Delta Lake și implementarea straturilor Medallion (Bronze, Silver, Gold).

sql_and_tuning 3 comenzi

core 3 comenzi

dataframe 13 comenzi

functions 11 comenzi

performance 5 comenzi

window 4 comenzi

actions 3 comenzi

udf 2 comenzi

streaming 4 comenzi

cli 3 comenzi

Toate funcțiile, comenzile și clasele Apache Spark & PySpark

Ghidul complet de referință pentru calcul distribuit și procesare Big Data cu Apache Spark 3.5 & PySpark. Consultați tabelele tematice de mai jos pentru a accesa paginile dedicate fiecărei metode, transformări și comenzi, însoțite de explicații tehnice clare și exemple practice de cod în limba română.

1. Sesiune Spark, Configurare & RDD-uri Fundamentale

Inițializarea clusterului, alocarea resurselor de memorie și operațiunile distribuite primitive pe RDD-uri.

Metodă / Comandă / Fațadă	Rol & Explicație Detaliată
`RDD & Catalyst Optimizer`	Arhitectura internă Apache Spark: trecerea de la RDD-uri la DataFrames, planuri de execuție logice și fizice cu optimizatorul Catalyst.
`SparkSession.builder`	Punctul principal de intrare în API-urile Spark SQL și DataFrame pentru configurarea aplicației.
`spark.conf.set()`	Ajustarea dinamică a parametrilor de memorie, shuffle partitions și optimizări la runtime.
`SparkContext & RDD`	Operații primitive de nivel scăzut pe structuri Resilient Distributed Datasets (sc.parallelize, map, reduceByKey).

2. Ingestie & Salvare Date (Data Source API)

Citirea și persistența distribuită a datelor în formate Parquet, CSV, JSON, Delta Lake, JDBC și tabele gestionate.

Metodă / Comandă / Fațadă	Rol & Explicație Detaliată
`spark.read.parquet`	Încărcarea eficientă a fișierelor columnare comprimate Parquet cu citire selectivă a coloanelor.
`spark.read.csv()`	Parsarea și ingestia paralelă a fișierelor CSV cu deducere de schemă sau StructType definit.
`spark.read.json()`	Încărcarea documentelor semi-structurate JSON și inferența automată a structurilor imbricate.
`Format Delta Lake`	Stocarea ACID cu versionare time-travel, garanții tranzacționale și suport upsert.
`spark.read.jdbc()`	Extragerea și încărcarea paralelă a datelor din baze relaționale SQL prin conexiuni JDBC multi-partitionate.
`df.write.mode()`	Configurarea comportamentului de persistență: append, overwrite, ignore sau errorIfExists.
`df.write.saveAsTable()`	Înregistrarea DataFrame-ului ca tabel gestionat în catalogul de date Hive Metastore sau Unity Catalog.
`partitionBy() & bucketBy()`	Organizarea fizică pe directoare ierarhice și hash buckets pentru accelerarea masivă a căutărilor.

3. Transformări DataFrame & Curățare Date

Selectarea, filtrarea, curățarea valorilor lipsă, deduplicarea și îmbinarea structurilor de date.

Metodă / Comandă / Fațadă	Rol & Explicație Detaliată
`Transformări & Acțiuni PySpark`	Procesare scalabilă cu PySpark: transformări narrow/wide (map, filter, join), acțiuni de declanșare a calculului și evaluare leneșă.
`df.select() & col()`	Selectarea, proiectarea și restrângerea coloanelor necesare din DataFrame.
`df.filter() & where()`	Filtrarea rândurilor pe baza condițiilor booleene simple sau compuse.
`df.withColumn()`	Adăugarea de coloane noi calculate sau actualizarea valorilor din coloanele existente.
`df.withColumnRenamed()`	Schimbarea numelor coloanelor fără recalcularea datelor sau modificarea tipurilor.
`df.drop()`	Eliminarea coloanelor redundante sau a datelor sensibile din schema DataFrame-ului.
`df.distinct()`	Eliminarea duplicatelor la nivelul întregului rând prin agregare hash distribuită.
`df.dropDuplicates()`	Deduplicarea înregistrărilor pe baza unui subset specificat de coloane cheie.
`df.na.fill() & df.na.drop()`	Tratarea valorilor nule prin imputare automată cu constante sau eliminare de rânduri.
`df.na.replace()`	Substituirea valorilor țintă pe baza unui dicționar de mapare de echivalențe.
`df.union() & unionByName()`	Concatenarea verticală a două tabele cu aliniere pozițională sau pe baza denumirilor coloanelor.
`df.sample()`	Extragerea unui eșantion statistic reprezentativ pentru experimente rapide și antrenare ML.
`df.cube() & df.rollup()`	Generarea automată a subtotalurilor și a agregărilor multidimensionale ierarhice.

4. Funcții Scalare PySpark SQL

Funcții native de expresii pentru manipularea textelor, numerelor, datelor temporale și a structurilor complexe.

Metodă / Comandă / Fațadă	Rol & Explicație Detaliată
`F.when().otherwise()`	Logică condițională CASE WHEN ramificată pentru crearea categoriilor derivate.
`F.lit()`	Crearea unui obiect Column constant dintr-o valoare scalară Python.
`F.concat() & F.concat_ws()`	Îmbinarea coloanelor de text cu separator ignorând automat valorile nule.
`F.split() & F.explode()`	Divizarea textelor în array-uri și despachetarea listelor în rânduri individuale.
`F.array_contains()`	Verificarea rapidă a existenței unei valori în colecții de tip listă.
`F.struct()`	Gruparea mai multor coloane individuale într-o structură compusă StructType.
`F.to_date() & to_timestamp()`	Parsarea șirurilor temporale și adăugarea/scăderea intervalelor calendaristice.
`F.datediff() & months_between()`	Calcularea numărului de zile și luni scurse între două date calendaristice.
`F.regexp_replace() & extract()`	Curățarea și extragerea de informații din texte pe baza tiparelor RegEx.
`F.coalesce()`	Selectarea primei valori non-NULL dintr-o serie de coloane evaluate succesiv.
`F.broadcast()`	Forțarea replicării tabelelor mici pe executori pentru eliminarea shuffling-ului la join.
`F.monotonically_increasing_id()`	Generarea de identificatori numerici unici la nivel de cluster pe 64 de biți fără blocare.
`df.join() & Broadcast Join`	Îmbinarea tabelelor distribuite prin inner, left, right, outer sau cross join.

5. Funcții Window & Analitice Avansate

Calcule pe partiții logice de date: clasamente, valori anterioare/următoare și medii mobile glisante.

Metodă / Comandă / Fațadă	Rol & Explicație Detaliată
`Window & F.row_number()`	Numerotarea secvențială a rândurilor per partiție pentru deduplicare sau top N elemente.
`F.rank() & F.dense_rank()`	Calcularea clasamentelor cu păstrarea sau eliminarea golurilor la scoruri egale.
`F.lead() & F.lag()`	Accesarea valorilor din rândurile imediat următoare sau anterioare fără auto-join.
`rowsBetween() Sume mobile`	Calculul sumelor cumulative și al mediilor pe ferestre glisante relative.

6. Acțiuni DataFrame, Inspecție & Statistică

Declanșarea execuției, afișarea rezultatelor, statistici agregate și inspecția schemei.

Metodă / Comandă / Fațadă	Rol & Explicație Detaliată
`df.show()`	Afișarea în consolă a primelor n rânduri ale DataFrame-ului în format ASCII structurat.
`df.printSchema()`	Vizualizarea ierarhică a arborelui de tipuri de date, coloanelor și nulabilității.
`df.count()`	Acțiune distribuită care numără numărul total de înregistrări din întregul DataFrame.
`df.collect() vs df.take()`	Aducerea rezultatelor în procesul driver și prevenirea erorilor OutOfMemory.
`df.describe() & df.summary()`	Generarea rapidă a indicatorilor statistici: medie, deviație, percentile, min și max.
`df.first() & df.head()`	Preluarea primului rând din DataFrame sub formă de obiect Row nativ.
`df.explain()`	Afișarea planurilor de execuție: Parsed, Analyzed, Optimized Logical și Physical Plan.

7. Performanță, Caching & Agregări Avansate

Gruparea datelor, pivotarea, optimizarea memoriei RAM prin cache și ruperea grafului de calcul.

Metodă / Comandă / Fațadă	Rol & Explicație Detaliată
`Tuning & Partitioning`	Tuning avansat în Spark SQL: strategii de partiționare, bucketing, broadcast hash joins și minimizarea shuffle-ului între noduri.
`df.groupBy() & agg()`	Gruparea datelor pe coloane cheie și aplicarea funcțiilor de agregare sum, avg, count.
`df.groupBy().pivot()`	Transpunerea valorilor categoriale dintr-o coloană în coloane orizontale distincte.
`df.crossJoin()`	Generarea produsului cartezian complet între două DataFrame-uri.
`df.orderBy() & sort()`	Sortarea globală distribuită a rândurilor în ordine crescătoare sau descrescătoare.
`repartition() vs coalesce()`	Optimizarea numărului de partiții prin redistribuire shuffle sau comasare locală.
`df.cache() & df.persist()`	Memorarea intermediară a DataFrame-ului în memoria RAM a executorilor pentru reutilizare rapidă.
`df.unpersist()`	Eliberarea manuală a memoriei RAM ocupate de DataFrame-urile salvate în cache.
`df.checkpoint()`	Salvarea fizică pe disc a stării și trunchierea arborelui genealogic de dependențe.

8. Optimizare Catalyst, UDF, Streaming & CLI

Adaptive Query Execution, funcții Python scalare și vectorizate Arrow, procesare de streaming și comenzi CLI.

Metodă / Comandă / Fațadă	Rol & Explicație Detaliată
`Structured Streaming & Delta Lake`	Ingestie și streaming continuu: conectare cu clustere Kafka, procesare micro-batch/continuous și arhitectura Medallion pe Delta Lake.
`Adaptive Query Execution (AQE)`	Re-optimizarea automată a planurilor de execuție la runtime pe baza statisticilor reale.
`F.udf() Scalar Python`	Scrierea de funcții personalizate în cod Python nativ cu specificarea tipului returnat.
`pandas_udf (Arrow Vectorized)`	Executarea calculelor vectorizate de mare viteză pe blocuri de memorie Apache Arrow.
`createOrReplaceTempView()`	Înregistrarea DataFrame-ului ca vedere temporară pentru executarea de interogări Spark SQL pure.
`spark.readStream Kafka`	Ingestia în timp real și consumul de fluxuri continue de evenimente din topic-uri Kafka.
`df.writeStream Delta / Sink`	Persistența continuă a datelor din flux în destinații Delta Lake cu recuperare la defectare.
`writeStream.trigger()`	Configurarea declanșării micro-batch-urilor la intervale fixe sau procesare în lot unic availableNow.
`df.withWatermark()`	Gestiunea datelor sosite cu întârziere și curățarea automată a stării din memorie.
`spark-submit CLI`	Utilitarul oficial de linie de comandă pentru trimiterea aplicațiilor pe clustere K8s sau YARN.
`pyspark & spark-shell`	Consola interactivă REPL pentru explorarea rapidă a datelor și prototipare.
`Spark UI (port 4040)`	Diagnosticarea grafică a stagiilor, identificarea blocajelor și analiza fenomenului de Data Skew.

Comunitate & Schimb de cunoștințe

Discuții, Întrebări & Coduri Apache Spark & PySpark

Ai o întrebare legată de Apache Spark & PySpark, cauți o soluție sau vrei să împărtășești un fragment util de cod? Trimite mai jos contribuția ta.

💬

Fii primul care pune o întrebare sau distribuie un fragment de cod / prompt!

Nu au fost adăugate încă întrebări sau exemple de cod pentru această pagină. Dacă ai întâlnit o problemă sau ai un snippet util, adaugă-l mai jos pentru a-i ajuta pe ceilalți.

Comunitate Adaugă o întrebare sau distribuie un cod Anulează răspunsul

Alege ce dorești să transmiți:

Titlu scurt (opțional)

Întrebarea ta *

Moderare & Siguranță:

Pentru a menține calitatea și securitatea informațiilor, fiecare întrebare sau fragment de cod/prompt este analizat (manual sau prin moderare asistată de AI) înainte de publicare.

Nume sau Pseudonim *

Adresă de Email * (nu va fi publicată)