Tutorial PySpark: RDD-uri, DataFrame-uri și Spark SQL în Python

Ce este PySpark?

PySpark este API-ul Python pentru Apache Spark. Codul tău Python descrie transformările dorite, iar motorul Spark, care rulează pe Java Virtual Machine (JVM), le planifică și le execută în întregul cluster. Obții execuția distribuită a lui Spark fără să scrii Scala sau Java.

Data scientist-ii folosesc PySpark pentru a curăța și agrega seturi mari de date, data engineer-ii îl folosesc pentru pipeline-uri ETL, iar ambele categorii folosesc pyspark.ml pentru a antrena modele pe date care nu încap pe o singură mașină. Dacă îți petreci majoritatea timpului în pandas, API-ul DataFrame îți va părea familiar.

Când să folosești PySpark în loc de pandas

PySpark merită când datele depășesc o singură mașină; sub acest prag, pandas este de obicei mai rapid deoarece nu are costuri de pornire JVM sau de programare a task-urilor.

Regula mea practică: dacă datele încap în memorie cu marjă și sarcina rulează pe un singur laptop, rămâi la pandas. Treci la PySpark când te lupți cu erori de memorie sau când sarcina va rula în cele din urmă pe un cluster sau pe o platformă precum Databricks.

Cum instalezi PySpark

PySpark se instalează din PyPI cu o singură comandă pip, care include și runtime-ul Spark, astfel încât nu ai nevoie de un download separat de Spark sau Hadoop pentru lucru local. Înainte să o rulezi, verifică cele două prerechizite.

Prerechizite

Conform ghidului oficial de instalare, PySpark 4.3 necesită:

• Python 3.11 sau mai nou. PySpark 4.3 a renunțat la Python 3.10; vezi ghidul nostru de configurare Python dacă ai nevoie de o versiune mai nouă.
• Java 17 sau mai nou (Spark 4.3 rulează pe Java 17, 21 și 25) cu variabila de mediu JAVA_HOME setată către aceasta. Ghidul nostru pentru instalarea Java explică pașii pentru fiecare sistem de operare.

Poți verifica ambele dintr-un terminal:

python3 --version  # Python 3.11 or later
java -version      # openjdk version "17" or laterFuncționează cu

Dacă lucrezi în DataLab sau Databricks, PySpark este deja instalat și poți trece direct la setul de date.

Instalarea PySpark cu pip

Îți recomand un mediu virtual ca să eviți coliziuni între dependențele PySpark și alte proiecte:

python3 -m venv pyspark-env
source pyspark-env/bin/activate
pip install "pyspark[sql]" "pandas<3"
python3 -c "import pyspark; print(pyspark.__version__)"Funcționează cu

4.3.0Funcționează cu

Pe Windows, folosește python în loc de python3 și activează mediul cu pyspark-env\Scripts\activate. Extra-ul [sql] adaugă pandas, PyArrow și NumPy, pe care PySpark le folosește pentru toPandas() și pandas UDF-uri. Fixez pandas<3 deoarece PySpark 4.3 afișează un avertisment că încă nu suportă complet pandas 3.0.

Două erori explică majoritatea eșecurilor la prima rulare:

• JAVA_GATEWAY_EXITED sau "Unable to locate a Java Runtime": Java lipsește sau JAVA_HOME nu este setată.
• PYTHON_VERSION_MISMATCH: Procesele worker ale Spark au preluat un alt Python decât scriptul tău. Setează PYSPARK_PYTHON către interpretul mediului tău, de exemplu export PYSPARK_PYTHON=$(which python).

Pentru pași specifici sistemului de operare, inclusiv Windows, vezi ghidul nostru de instalare PySpark.

Obținerea setului de date

Exemplele folosesc un set de date e-commerce cu 541.909 tranzacții ale unui retailer online din Marea Britanie, acoperind perioada decembrie 2010 – decembrie 2011. Fiecare rând reprezintă o linie de produs pe o factură, cu 8 coloane:

• InvoiceNo: numărul facturii (numerele care încep cu "C" sunt anulări)
• StockCode și Description: codul și numele produsului
• Quantity și UnitPrice: unități cumpărate și preț pe unitate, în lire sterline
• InvoiceDate: data și ora achiziției
• CustomerID: identificatorul clientului (lipsește pe unele rânduri)
• Country: țara clientului

Îl poți descărca ca datacamp_ecommerce.csv de pe pagina setului de date DataLab. Este setul de date Online Retail din UCI Machine Learning Repository, astfel încât poți construi același fișier și din sursa publică (necesită pip install openpyxl):

import io
import urllib.request
import zipfile

import pandas as pd

url = "https://archive.ics.uci.edu/static/public/352/online+retail.zip"
with urllib.request.urlopen(url) as response:
    archive = zipfile.ZipFile(io.BytesIO(response.read()))

retail = pd.read_excel(
    archive.open("Online Retail.xlsx"),
    dtype={"InvoiceNo": str, "StockCode": str, "CustomerID": "Int64"},
)
# Match the DataLab file's date format, e.g. 12/1/10 8:26
retail["InvoiceDate"] = retail["InvoiceDate"].map(
    lambda ts: f"{ts.month}/{ts.day}/{ts:%y} {ts.hour}:{ts:%M}"
)
retail.to_csv("datacamp_ecommerce.csv", index=False)Funcționează cu

Citirea fișierului Excel de 23 MB durează aproximativ un minut. Salvează CSV-ul în folderul din care vei rula restul codului.

Crearea unui SparkSession

Un SparkSession este punctul de intrare pentru orice program PySpark: leagă codul tău Python de motorul Spark și creează DataFrame-uri. Creezi unul per aplicație, folosind builder pattern:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("PySparkTutorial").getOrCreate()
print(spark.version)Funcționează cu

4.3.0Funcționează cu

getOrCreate() returnează sesiunea deja pornită dacă există una, deci e sigur să o apelezi de două ori într-un notebook. Lansat dintr-un proces Python obișnuit, Spark rulează în modul local (local[*]), folosind fiecare nucleu CPU al mașinii tale ca worker, și servește interfața Spark la http://localhost:4040 pentru monitorizarea joburilor. Când ai terminat, spark.stop() eliberează resursele.

Încărcarea setului de date e-commerce

Cu sesiunea pornită, putem citi fișierul CSV într-un DataFrame. Trec o schemă explicită în loc de inferSchema=True, care face ca Spark să scaneze încă o dată întregul fișier doar pentru a ghici tipurile coloanelor:

import pyspark.sql.functions as F

schema = """
    InvoiceNo STRING, StockCode STRING, Description STRING, Quantity INT,
    InvoiceDate STRING, UnitPrice DOUBLE, CustomerID INT, Country STRING
"""

orders = spark.read.csv(
    "datacamp_ecommerce.csv", header=True, schema=schema, escape='"'
)
orders = orders.withColumn("InvoiceDate", F.to_timestamp("InvoiceDate", "M/d/yy H:mm"))

orders.printSchema()Funcționează cu

root
 |-- InvoiceNo: string (nullable = true)
 |-- StockCode: string (nullable = true)
 |-- Description: string (nullable = true)
 |-- Quantity: integer (nullable = true)
 |-- InvoiceDate: timestamp (nullable = true)
 |-- UnitPrice: double (nullable = true)
 |-- CustomerID: integer (nullable = true)
 |-- Country: string (nullable = true)Funcționează cu

Două detalii din fragmentul de mai sus contează:

• escape='"' gestionează numele de produse care conțin ghilimele duble, cum ar fi RECORD FRAME 7"" SINGLE SIZE. Caracterul de escape implicit al Spark este backslash.
• Modelul de dată M/d/yy H:mm se potrivește cu valori precum 12/1/10 8:26. Începând cu Spark 4.0, modul ANSI SQL este activ implicit, așa că to_timestamp() ridică o eroare CANNOT_PARSE_TIMESTAMP pentru o valoare care nu se potrivește cu modelul. Folosește F.try_to_timestamp() dacă preferi să obții NULL pentru rândurile invalide.

Pentru a vedea primele rânduri și a le număra:

orders.show(5, truncate=False)
print(orders.count())Funcționează cu

+---------+---------+-----------------------------------+--------+-------------------+---------+----------+--------------+
|InvoiceNo|StockCode|Description                        |Quantity|InvoiceDate        |UnitPrice|CustomerID|Country       |
+---------+---------+-----------------------------------+--------+-------------------+---------+----------+--------------+
|536365   |85123A   |WHITE HANGING HEART T-LIGHT HOLDER |6       |2010-12-01 08:26:00|2.55     |17850     |United Kingdom|
|536365   |71053    |WHITE METAL LANTERN                |6       |2010-12-01 08:26:00|3.39     |17850     |United Kingdom|
|536365   |84406B   |CREAM CUPID HEARTS COAT HANGER     |8       |2010-12-01 08:26:00|2.75     |17850     |United Kingdom|
|536365   |84029G   |KNITTED UNION FLAG HOT WATER BOTTLE|6       |2010-12-01 08:26:00|3.39     |17850     |United Kingdom|
|536365   |84029E   |RED WOOLLY HOTTIE WHITE HEART.     |6       |2010-12-01 08:26:00|3.39     |17850     |United Kingdom|
+---------+---------+-----------------------------------+--------+-------------------+---------+----------+--------------+
only showing top 5 rows

541909Funcționează cu

Pentru mai multe opțiuni de citire, precum fișiere multiple, delimitatori și înregistrări corupte, vezi tutorialul nostru PySpark pentru citirea CSV.

RDD-uri vs. DataFrame-uri

RDD-urile și DataFrame-urile sunt cele două modalități prin care PySpark reprezintă date distribuite, iar pentru aproape tot codul nou, DataFrame-urile sunt alegerea potrivită.

Iată aceeași agregare, cantitatea totală pe țară, scrisă în ambele moduri:

sales = [("United Kingdom", 6), ("France", 24), ("United Kingdom", 8)]

# RDD: scrii logica cu funcții Python
rdd = spark.sparkContext.parallelize(sales)
print(rdd.reduceByKey(lambda a, b: a + b).collect())

# DataFrame: declari rezultatul, iar Spark plănuiește cum să-l calculeze
sales_df = spark.createDataFrame(sales, ["country", "quantity"])
sales_df.groupBy("country").sum("quantity").show()Funcționează cu

[('France', 24), ('United Kingdom', 14)]
+--------------+-------------+
|       country|sum(quantity)|
+--------------+-------------+
|United Kingdom|           14|
|        France|           24|
+--------------+-------------+Funcționează cu

Versiunea RDD rulează lambda-ul tău în procese worker Python. Versiunea DataFrame trece prin optimizerul Catalyst al Spark, care rescrie interogarea într-un plan eficient ce rulează în JVM, așa că de obicei codul pe DataFrame este mai rapid fără reglaje. Restul tutorialului folosește DataFrame-uri.

Evaluare întârziată: Transformări vs. Acțiuni

Evaluarea întârziată înseamnă că Spark nu execută o transformare când o apelezi; înregistrează pasul într-un plan și rulează întregul plan doar când o acțiune cere un rezultat.

Filtrul de mai jos returnează instant, chiar dacă setul de date are peste jumătate de milion de rânduri, pentru că încă nu s-a citit nimic. Apelul count() este cel care lansează un job:

uk_orders = orders.filter(F.col("Country") == "United Kingdom")  # transformation: no job yet
print(uk_orders.count())                                          # action: Spark reads the file nowFuncționează cu

495478Funcționează cu

Așteptarea până la o acțiune îi permite lui Spark să optimizeze întregul lanț dintr-odată. explain() afișează planul fizic ales (scurtat aici):

uk_orders.explain()Funcționează cu

== Physical Plan ==
*(1) Project [InvoiceNo#0, StockCode#1, ...]
+- *(1) Filter (isnotnull(Country#7) AND (Country#7 = United Kingdom))
   +- FileScan csv [...] PushedFilters: [IsNotNull(Country), EqualTo(Country,United Kingdom)], ...Funcționează cu

Intrarea PushedFilters arată cum Spark împinge filtrul pe țară în citirea fișierului, astfel încât rândurile din alte țări sunt eliminate pe măsură ce CSV-ul este citit. O consecință practică: o eroare de date într-o transformare, cum ar fi o dată care nu poate fi parsată în modul ANSI, apare abia când rulează următoarea acțiune, câteva linii mai jos.

Operațiuni uzuale pe DataFrame

Majoritatea codului PySpark este un lanț din cinci operațiuni pe DataFrame: select(), filter(), withColumn(), groupBy() și join(). Fiecare returnează un nou DataFrame, deci le poți înlănțui în orice ordine.

Selectarea coloanelor cu select()

select() returnează un DataFrame doar cu coloanele pe care le numești:

orders.select("InvoiceNo", "Description", "Quantity", "UnitPrice").show(3, truncate=False)Funcționează cu

+---------+----------------------------------+--------+---------+
|InvoiceNo|Description                       |Quantity|UnitPrice|
+---------+----------------------------------+--------+---------+
|536365   |WHITE HANGING HEART T-LIGHT HOLDER|6       |2.55     |
|536365   |WHITE METAL LANTERN               |6       |3.39     |
|536365   |CREAM CUPID HEARTS COAT HANGER    |8       |2.75     |
+---------+----------------------------------+--------+---------+
only showing top 3 rowsFuncționează cu

Filtrarea rândurilor cu filter()

filter() păstrează rândurile care îndeplinesc o condiție. Acest set de date amestecă vânzări cu anulări (numere de factură care încep cu „C”), cantități negative și rânduri fără client, așa că voi păstra doar vânzările valide:

sales_only = orders.filter(
    (F.col("Quantity") > 0)
    & (F.col("UnitPrice") > 0)
    & F.col("CustomerID").isNotNull()
    & ~F.col("InvoiceNo").startswith("C")
)
print(sales_only.count())Funcționează cu

397884Funcționează cu

Combină condițiile cu & (și), | (sau) și ~ (nu) și pune fiecare condiție între paranteze, fiindcă altfel ordinea operatorilor din Python ar evalua & înainte de >. Tutorialul nostru PySpark pentru filter acoperă mai multe tipare.

Adăugarea de coloane cu withColumn()

withColumn() adaugă o coloană sau o înlocuiește pe una cu același nume. Aici calculează valoarea fiecărei linii:

sales_only = sales_only.withColumn("TotalAmount", F.round(F.col("Quantity") * F.col("UnitPrice"), 2))
sales_only.select("InvoiceNo", "Quantity", "UnitPrice", "TotalAmount").show(3)Funcționează cu

+---------+--------+---------+-----------+
|InvoiceNo|Quantity|UnitPrice|TotalAmount|
+---------+--------+---------+-----------+
|   536365|       6|     2.55|       15.3|
|   536365|       6|     3.39|      20.34|
|   536365|       8|     2.75|       22.0|
+---------+--------+---------+-----------+
only showing top 3 rowsFuncționează cu

Vezi tutorialul nostru PySpark withColumn() pentru coloane condiționale cu F.when().

Agregarea cu groupBy() și agg()

groupBy() grupează rândurile după una sau mai multe coloane, iar agg() calculează una sau mai multe sumarizări pe grup. Care țări aduc cele mai mari venituri?

country_revenue = (
    sales_only.groupBy("Country")
    .agg(
        F.count_distinct("CustomerID").alias("customers"),
        F.round(F.sum("TotalAmount"), 2).alias("revenue"),
    )
    .orderBy(F.desc("revenue"))
)
country_revenue.show(5)Funcționează cu

+--------------+---------+----------+
|       Country|customers|   revenue|
+--------------+---------+----------+
|United Kingdom|     3920|7308391.55|
|   Netherlands|        9| 285446.34|
|          EIRE|        3|  265545.9|
|       Germany|       94| 228867.14|
|        France|       87| 209024.05|
+--------------+---------+----------+
only showing top 5 rowsFuncționează cu

Regatul Unit domină, cum te-ai aștepta de la un retailer din Marea Britanie. Rândurile mai interesante sunt Țările de Jos și Irlanda (listată ca EIRE): 9 și 3 clienți cheltuiesc mai mult decât cei 94 din Germania, ceea ce sugerează cumpărători en-gros.

Observă că import modulul functions ca F în loc să folosesc from pyspark.sql.functions import *, care umbrește sum(), min() și max() din Python. Începând cu PySpark 4.0, acel import wildcard nu mai expune nici vechiul nume camelCase countDistinct.

Tutorialul nostru PySpark groupBy intră mai adânc în agregări.

Îmbinarea DataFrame-urilor cu join()

join() combină două DataFrame-uri pe o cheie comună. Pentru a eticheta fiecare țară cu o regiune, voi îmbina un tabel mic de lookup care omite deliberat Australia:

regions = spark.createDataFrame(
    [("United Kingdom", "UK"), ("Netherlands", "EU"), ("EIRE", "EU"),
     ("Germany", "EU"), ("France", "EU")],
    ["Country", "region"],
)
country_revenue.join(regions, on="Country", how="left").orderBy(F.desc("revenue")).show(6)Funcționează cu

+--------------+---------+----------+------+
|       Country|customers|   revenue|region|
+--------------+---------+----------+------+
|United Kingdom|     3920|7308391.55|    UK|
|   Netherlands|        9| 285446.34|    EU|
|          EIRE|        3|  265545.9|    EU|
|       Germany|       94| 228867.14|    EU|
|        France|       87| 209024.05|    EU|
|     Australia|        9| 138521.31|  NULL|
+--------------+---------+----------+------+
only showing top 6 rowsFuncționează cu

Un left join păstrează fiecare rând din DataFrame-ul din stânga și completează nepotrivirile cu NULL, motiv pentru care Australia rămâne în rezultat. Un inner join (how="inner", implicitul) ar fi eliminat-o. Tutorialul nostru despre join-uri în PySpark compară toate tipurile de join și performanța lor.

Spark SQL în PySpark

Spark SQL îți permite să interoghezi un DataFrame cu SQL standard, înregistrându-l ca vizualizare temporară. Această interogare reproduce tabelul cu veniturile pe țară din secțiunea groupBy:

sales_only.createOrReplaceTempView("sales")

spark.sql("""
    SELECT Country,
           COUNT(DISTINCT CustomerID) AS customers,
           ROUND(SUM(TotalAmount), 2) AS revenue
    FROM sales
    GROUP BY Country
    ORDER BY revenue DESC
    LIMIT 5
""").show()Funcționează cu

+--------------+---------+----------+
|       Country|customers|   revenue|
+--------------+---------+----------+
|United Kingdom|     3920|7308391.55|
|   Netherlands|        9| 285446.34|
|          EIRE|        3|  265545.9|
|       Germany|       94| 228867.14|
|        France|       87| 209024.05|
+--------------+---------+----------+Funcționează cu

Ambele versiuni se compilează în același plan optimizat, deci niciuna nu este mai rapidă. Folosesc SQL pentru agregări lungi pe care analiștii le vor citi, iar API-ul DataFrame când am nevoie de logică Python în jurul interogării, precum construirea de filtre într-o buclă. spark.sql() returnează un DataFrame, deci le poți combina liber.

Vizualizarea temporară durează până când se oprește SparkSession.

Machine Learning cu PySpark: Un exemplu K-Means

Modulul pyspark.ml al PySpark antrenează modele pe DataFrame-uri, iar un Pipeline leagă pregătirea caracteristicilor și modelul într-un singur obiect pe care îl poți antrena și reutiliza. Ca exemplu, voi segmenta clienții retailerului cu clustering K-Means pe recency, frequency și monetary value (RFM), o segmentare standard în marketing:

• Recency: zile de la ultima achiziție a clientului
• Frequency: numărul de facturi distincte
• Monetary: suma totală cheltuită

Construirea caracteristicilor RFM cu groupBy()

Un singur groupBy() peste DataFrame-ul curățat sales_only produce toate cele 3 caracteristici. Recency se măsoară din ziua de după ultima factură din date:

snapshot = sales_only.agg(F.date_add(F.max("InvoiceDate"), 1)).first()[0]

rfm = sales_only.groupBy("CustomerID").agg(
    F.datediff(F.lit(snapshot), F.max("InvoiceDate")).alias("recency"),
    F.count_distinct("InvoiceNo").alias("frequency"),
    F.round(F.sum("TotalAmount"), 2).alias("monetary"),
)
rfm.describe("recency", "frequency", "monetary").show()Funcționează cu

+-------+------------------+-----------------+------------------+
|summary|           recency|        frequency|          monetary|
+-------+------------------+-----------------+------------------+
|  count|              4338|             4338|              4338|
|   mean| 93.05947441217151|4.272014753342554|2054.2664591977896|
| stddev|100.01226424171826|7.697997826095038| 8989.230441421178|
|    min|                 1|                1|              3.75|
|    max|               374|              209|         280206.02|
+-------+------------------+-----------------+------------------+Funcționează cu

Uită-te la coloana monetary: o medie de aproximativ £2.054, o deviație standard de peste patru ori mai mare și un maxim peste £280.000. K-Means măsoară distanța euclidiană, deci câțiva mari cheltuitori ar domina clusterele. Luarea logaritmului fiecărei caracteristici comprimă acea coadă lungă.

Scalare și clustering într-un Pipeline

Pipeline-ul are 3 etape: VectorAssembler împachetează caracteristicile în singura coloană vector pe care o așteaptă modelele pyspark.ml, StandardScaler le aduce pe aceeași scară, iar KMeans le grupează. Îl antrenez pentru k = 2 până la 6 și punctez fiecare model cu silhouette score, care variază între -1 și 1, mai mare însemnând clustere mai bine separate:

from pyspark.ml import Pipeline
from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
from pyspark.ml.feature import StandardScaler, VectorAssembler

rfm_log = rfm.select(
    "CustomerID",
    "recency", "frequency", "monetary",
    F.log1p("recency").alias("log_recency"),
    F.log1p("frequency").alias("log_frequency"),
    F.log1p("monetary").alias("log_monetary"),
).cache()

assembler = VectorAssembler(
    inputCols=["log_recency", "log_frequency", "log_monetary"], outputCol="features_raw"
)
scaler = StandardScaler(
    inputCol="features_raw", outputCol="features", withMean=True, withStd=True
)
evaluator = ClusteringEvaluator(featuresCol="features", metricName="silhouette")

for k in range(2, 7):
    kmeans = KMeans(featuresCol="features", k=k, seed=42)
    model = Pipeline(stages=[assembler, scaler, kmeans]).fit(rfm_log)
    score = evaluator.evaluate(model.transform(rfm_log))
    print(f"k={k}  silhouette={score:.3f}")Funcționează cu

k=2  silhouette=0.624
k=3  silhouette=0.502
k=4  silhouette=0.502
k=5  silhouette=0.478
k=6  silhouette=0.470Funcționează cu

.cache() păstrează rfm_log în memorie, pentru că fiecare antrenare și evaluare din buclă îl citește din nou; fără asta, Spark ar recomputa tabelul RFM din CSV la fiecare antrenare.

Alegerea lui k și interpretarea segmentelor

Scorul silhouette atinge vârful la k = 2, dar 2 segmente separă în principal clienții activi de cei inactivi, cu care echipa de marketing poate face puține. k = 4 obține același scor ca k = 3 și împarte clienții activi în mari cheltuitori și nou-veniți, așa că îl voi folosi și voi profila fiecare cluster:

kmeans = KMeans(featuresCol="features", k=4, seed=42)
model = Pipeline(stages=[assembler, scaler, kmeans]).fit(rfm_log)
segments = model.transform(rfm_log)

(segments.groupBy("prediction")
    .agg(
        F.count("*").alias("customers"),
        F.round(F.avg("recency"), 1).alias("avg_recency"),
        F.round(F.avg("frequency"), 1).alias("avg_frequency"),
        F.round(F.avg("monetary"), 1).alias("avg_monetary"),
    )
    .orderBy("prediction")
    .show())Funcționează cu

+----------+---------+-----------+-------------+------------+
|prediction|customers|avg_recency|avg_frequency|avg_monetary|
+----------+---------+-----------+-------------+------------+
|         0|     1598|      181.2|          1.3|       326.2|
|         1|      720|       13.2|         13.7|      8072.7|
|         2|     1162|       77.1|          3.9|      1765.2|
|         3|      858|       17.4|          2.3|       613.7|
+----------+---------+-----------+-------------+------------+Funcționează cu

Cele patru segmente se interpretează clar:

• Clusterul 1, cei mai buni clienți (720): au cumpărat în ultimele 2 săptămâni în medie, cu aproximativ 14 comenzi și peste £8.000 cheltuiți fiecare. Aici se potrivesc ofertele de retenție și loialitate.
• Clusterul 3, recenți și cu volum mic (858): activi în ultimele 3 săptămâni, dar cu doar vreo 2 comenzi. Mulți sunt probabil clienți noi care merită îndrumați spre a doua și a treia comandă.
• Clusterul 2, în răcire (1.162): cheltuieli medii, dar cam 11 săptămâni de la ultima achiziție. O campanie de recâștigare are cele mai bune șanse aici.
• Clusterul 0, cumpărători ocazionali care au dispărut (1.598): în mare parte o singură comandă, în jur de £326, acum aproximativ 6 luni. Cel mai mare grup și cel mai puțin valoros.

Numerele clusterelor sunt etichete arbitrare, iar o altă sămânță le poate reordona, așa că profilează întotdeauna clusterele înainte de a le numi. Pentru a înțelege cum își alege K-Means centroizii, vezi introducerea noastră în clusteringul k-means, iar referința API pyspark.ml listează fiecare model și transformator. Cursul nostru Machine Learning cu PySpark acoperă clasificare, regresie și reglarea modelelor cu Pipeline-uri.

Gânduri de final

PySpark se reduce la câteva idei: un SparkSession pentru început, DataFrame-uri pentru aproape tot, transformări leneșe care rulează doar când o acțiune are nevoie de un rezultat și alegerea de a scrie fiecare interogare în Python sau SQL. Tot ce este în acest tutorial a rulat în modul local pe un laptop, iar același cod rulează neschimbat pe un cluster.

Un fișier cu 541.909 rânduri nu are nevoie de un cluster, iar pandas s-ar fi descurcat. Asta a fost intenționat, pentru că obiceiurile care contează pe date mari (scheme explicite, filtrare devreme, caching pentru ceea ce refolosești și verificarea cu explain()) sunt mai ușor de format pe date pe care le poți inspecta.

De aici, acestea sunt următorii pași buni, în funcție de obiectivul tău:

• Exersează API-ul DataFrame cu exerciții: Introducere în PySpark
• Pregătește date reale, dezordonate: Curățarea datelor cu PySpark
• Păstrează sintaxa la îndemână: fișa noastră PySpark
• Pregătește-te pentru un interviu de angajare: Top întrebări de interviu PySpark

Dacă vrei să mergi mai departe, îți recomand parcursul Big Data cu PySpark, care acoperă gestionarea datelor, ingineria caracteristicilor, pipeline-uri de machine learning și motoare de recomandare, în ordine. Pentru echipe, DataCamp for Business oferă trasee de învățare pentru roluri de data engineering.