I-PySpark yabaqalayo: Ngale Kwezisekelo

kulolu chungechunge, I-PySpark yabaqalayo: Ukuqonda Okuyisisekelo lapho-ke usuvele uqonda inhliziyo ye-Spark: idatha esabalalisiwe, ama-DataFrames, nokubulawa okuvilaphayo. Ufake i-PySpark, wamisa i-SparkSession, wafunda i-CSV futhi wenza ukukhohlisa kwedatha ku-Dataframe. Ngizoshiya isixhumanisi saleyo ndaba ekugcineni kwalena.

Into eyodwa okufanele iphinde iphindwe kuleso sihloko sokuqala ukuthi ngivame ukusebenzisa amagama e-PySpark kanye ne-Spark ngokushintshana, kodwa uma ngikhuluma ngokuqinile, i-Spark wuhlaka lwekhompiyutha olusabalalisiwe olubanzi (olubhalwe ku-Scala), kanti i-PySpark iyi-API ye-Python ezinikele ku-Spark.

Ngale kwezisekelo

Manje, kukhona okuthakazelisayo okwenzekayo uma udlula kuleso sigaba sabasaqalayo. Uyabona ngokushesha eyakho okwesibili Iphrojekthi ye-PySpark idinga umqondo ohluke kancane:

• Ufuna ukufunda/ukubhala idatha ngendlela ephephile, esheshayo, nengabikezelwa.
• Ufuna ukuhlanganisa amasethi edatha ngaphandle kokuzizwa ungaqiniseki ngokujoyina.
• Ufuna ukuqonda ngani I-Spark iziphatha ngendlela eyenza ngayo – nokuthi ungayigudluzela kanjani ngobumnene iqonde endaweni efanele.

Lesi sihloko sikuhambisa kulezo zinyathelo ezilandelayo. Ihamba kancane ngamabomu futhi iyasebenza. Abekho abangaphakathi abajulile. Akukho ukushuna kweqoqo. Akukho ukulungiselelwa kwe-Spark okuyinkimbinkimbi. Izinto nje abasaqalayo bangempela okudingeka bazazi lapho besuka ezibonelweni zamathoyizi baye emsebenzini omncane, womhlaba wangempela.

Sisebenzisa i-open-source Spark, esebenza endaweni, njengangaphambili.

1. Ukuthatha isinyathelo esilandelayo: ukufunda idatha ngendlela efanele

Esihlokweni sami sokuqala, sisebenzise isilayishi se-CSV esilula kakhulu:

df = spark.read.csv("sales.csv", header=True, inferSchema=True)

Iyasebenza – futhi ilungele ukuhlolwa kwasekuqaleni – kodwa ifihla inkinga ecashile.

U-Spark uqagela izinhlobo zedatha yakho

Uma usebenzisa i-inferSchema=True Directive, uSpark ubheka isampula encane yefayela lakho bese esebenzisa lolo lwazi ukuze qagela noma ngabe ikholomu iyinani eliphelele, iyunithi yezinhlamvu, i-boolean, noma i-double. Lokho kusho ukuthi:

• Uma imigqa engu-99 ibonakala iyinombolo futhi umugqa we-100 ungenalutho, i-Spark ingase ihumushe ikholomu njenge umucu.
• Uma othile ehlela ifayela ngesonto elizayo futhi ngephutha engeza u-£23.50 esikhundleni sika-23.50, uSpark angase ayiphathe ngendlela ehlukile ikholomu yonke.
• Uma ifayela lakho likhulu, isampula esetshenziswa yi-Spark ngeke imele yonke idathasethi.

Lokhu kungaholela ekuziphatheni okungaqondakali kamuva, uhlobo lwezinambuzane abaqalayo abakuthola kunzima kakhulu ukulixilonga.

Umkhuba ongcono wabaqalayo: chaza i-schema sedatha yakho

Cabanga nge-schema njengenguqulo ye-Spark yepulani yokufunda idatha. Ngaphambi kokwakha noma yini, utshela uSpark izinto ezifana nokuthi:

Amagama amakholomu
Iluphi uhlobo lwedatha okufanele lube
Ukuthi inani lekholomu lingokuzithandela.

Nakhu ukuthi kubukeka kanjani ngesibonelo sethu sedatha yokuthengisa. Khumbula ukuthi idatha ibibukeka kanje:

transaction_id,customer_name,net_amount,tax_amount, is_member
101,Alice,250.50,25.05,true
102,Bob,120.00,6.00, false
103,Charlie,450.75,25.07,true
104,David,89.99,5.73,false

Ukucacisa izinhlobo zezinkambu ezingenhla ku-Spark, sichaza i-schema yethu sisebenzisa ikhodi efana nale elandelayo.

from pyspark.sql import types as T

schema = T.StructType([
T.StructField("transaction_id", T.IntegerType(), False),
T.StructField("customer_name", T.StringType(), False),
T.StructField("net_amount", T.DoubleType(), True),
T.StructField("tax_amount", T.DoubleType(), True),
T.StructField("is_member", T.BooleanType(), True),
])

Amagama ekholomu nezinhlaka zohlobo kuzichaza ngokwazo. Iqiniso[False] ipharamitha ibonisa ukuthi kungenzeka [not] ube NULL amanani kukholamu. Qaphela, ifulegi elithi Iqiniso/Amanga lokungabi nalutho ngokuvamile liyimethadatha ye-schema nolwazi lokuthuthukisa. Akuhlala kugcizelelwa ngokuqinile kuyo yonke imithombo yedatha ngendlela i-database NOT NULL constraindication.

Izinketho eziningi eziwusizo lapho ufunda idatha ye-CSV

Kunenqwaba yezinketho eziwusizo zokufunda ze-CSV ongazihlanganisa nomyalelo we-schema okwenza ukulayisha idatha ye-CSV kuthembeke nakakhulu.

Izinketho ezivame kakhulu zihlanganisa:

• imodi=”IMVUME”: igcina amarowu amabi kakhulu ngangokunokwenzeka
• imodi=”DROPMALFORMED”: yehlisa imigqa engalungile
• mode=”FAILFAST”: amaphutha ngokushesha
• isihloko = Iqiniso[False]: Ingabe ifayela liqukethe [or not] irekhodi unhlokweni
• i-nullValue: imuphi umbhalo okufanele ungene esikhundleni samanani angenalutho kokokufakayo
• dateFormat / timestampFormat

Manje singakwazi ukulayisha idatha_yokuthengisa ku-Dataframe kanje:

df = (
    spark.read
    .option("header", True)
    # Other modes: "PERMISSIVE" and "DROPMALFORMED".
    .option("mode", "FAILFAST")
    .option("nullValue", "N/A")
    .schema(schema)
    .csv("sales_data.csv")
)

Kungani lokhu kubalulekile kwabaqalayo?

1. Wena yazi ukuthi ziyini izinhlobo zedatha ngaphambi kokuthi uqale ukusebenza.
2. Uma kucacisiwe, i-Spark izokwenza yenqaba imigqa eyinqaba esikhundleni sokuzihumusha buthule.
3. Izinguquko zakho ziba ziningi ukubikezelwa.
4. Uma ujoyina amasethi edatha amabili kamuva, thayipha ukungafani ngeke kukumangaze.

2. Ukuqonda ukuguqulwa kwedatha

Khumbula, esihlokweni sami sangaphambilini, ezinyathelweni zethu zokuqala zokukhohlisa ama-dataframes nge-PySpark, sengeze ikholomu eyengeziwe, esuselwe ku-Dataframe yethu sisebenzisa ikhodi enjengale:

df2 = df.withColumn("gross_amount", df.net_amount + df.tax_amount)

Ngachaza ukuthi lo mugqa awubale lutho okwamanje. Imane yengeze isinyathelo kuhlelo lwangaphakathi lwe-Spark:

1. Read the CSV  
2. Add a new column (gross_amount = net + tax)

Bese ungangeza izinyathelo ezinjengalezi:

df3 = df2.withColumn("tax_percentage", df2.tax_amount / df2.gross_amount * 100)

Noma kunjalo, akukho zibalo eziye zenzeka. Kuphela uma wenza i- isenzo njengo…

df3.show()

… ingabe u-Spark uthi:

“Kulungile, manje ngidinga ukugijima zonke lezi zinyathelo.”

Yilokhu okushiwo “ukubulala ngokuvilapha”, kodwa okubalulekile kwabaqalayo akulona igama. Kungumphumela, futhi kusho ukuthi,

• Ungakwazi ukuhlanganisa izinguquko eziningi ngaphandle “kokuzikhokhela” kuze kube yilapho udinga umphumela.
• I-Spark ingakwazi ukuhlela kabusha i-oda ngaphakathi ukuze iqhube izinto kahle.
• Awuchithi isikhathi ngokwenza izinyathelo ezimaphakathi kudatha ongase uyihlunge ngemva kwesikhathi.

Cabanga ngakho njengoba ungenza umsebenzi wansuku zonke, njengokwenza isemishi:

• Uqoqa zonke izithako.
• Uyakuhlanganisa engqondweni yakho.
• Empeleni uqala ukusika nokuzilungiselela uma wazi kahle ukuthi wenzani.

3. Ukuhlanza idatha ngaphambi kokuthi kubangele izinkinga

Idatha yangempela ngokuvamile ingcolile futhi ngokuvamile iqukethe amanani angekho, izintambo ezingenalutho, amarekhodi ayimpinda, noma amanani wesimeli njengokuthi “N/A” nokuthi “akwaziwa”.

Ku-PySpark, umgomo uwukubamba futhi ubhekane nezinkinga ezisobala kusenesikhathi ukuze konke ukuhamba kwakho komsebenzi kuziphathe ngendlela ebikezelwayo. I-PySpark inemisebenzi eminingi ewusizo eyenza ukwazi ukwenza lokhu.

Ilahla imigqa enamanani angekho

Umsebenzi wokuhlanza olula yi-dropna().

df_clean = df.dropna()

Lokhu kususa noma yimuphi umugqa oqukethe inani elingenalutho kunoma iyiphi ikholomu. Lokho kungaba usizo, kodwa kuvame ukuba nolaka kakhulu.

Ngokuvamile, uphonsa kuphela imigqa lapho amakholomu abalulekile kulowo mugqa engekho khona:

df_clean = df.dropna(subset=["net_amount", "tax_amount"])

Lokhu kusho:

Gcina umugqa inqobo nje uma inani_elilotshiwe kanye nenani_lentela kukhona.

Amanye amakholomu asengaqukatha ama-null, futhi lokho kungase kube kuhle.

Ukugcwalisa amanani angekho

Kwesinye isikhathi awufuni ukususa imigqa. Ufuna nje ukushintsha amanani angekho ngokuthile okunengqondo.

Kulapho i-fillna() iwusizo khona.

df_clean = df.fillna({"city": "Unknown"})

Ungaphinda ugcwalise amakholomu ezinombolo:

df_clean = df.fillna({"tax_amount": 0.0})

Lokhu kuyasiza uma inani elingekho linencazelo ecacile. Isibonelo, inani lesaphulelo elishodayo lingase libe ngu-0.0. Kodwa qaphela. Ukugcwalisa amanani angekho kungashintsha incazelo yedatha yakho uma ukhetha okumisiwe okungalungile.

Ukushintsha izinhlobo zekholomu ngohlaka()

Kwesinye isikhathi i-Spark ifunda ikholomu njengohlobo olungalungile, ikakhulukazi uma isebenza ngamafayela e-CSV. Uma kunjalo, ungakwazi ukuguqula ikholomu usebenzisa i-cast() opharetha:

from pyspark.sql import functions as F 
df_clean = df.withColumn("net_amount",F.col("net_amount").cast("double") )

Lokhu kuvame kakhulu lapho amadethi, izinombolo, noma ama-booleans afundwe njengeyunithi yezinhlamvu.

Ikhipha imigqa eyimpinda

Imigqa eyimpinda ingavela uma amafayela ethunyelwa ngaphezu kokukodwa, ehlanganiswe ngokungalungile, noma ehlanganiswe emithonjeni eminingi. Ungasusa izimpinda ezifana nalezi:

df_clean = df.dropDuplicates()

Noma susa izimpinda ngokusekelwe kukholomu eyodwa noma amaningi akhethiwe.

df_clean = df.dropDuplicates(["transaction_id"])

Leyo nguqulo yesibili ivame ukuba usizo kakhulu ngoba ithi:

I-ID ngayinye yokwenziwe kufanele ivele kanye kuphela.

Isibonelo esincane sokuhlanza idatha

Ukuhlanganisa leyo mibono:

from pyspark.sql import functions as F


df_clean = (
    df
    # Remove transactions missing required values.
    .dropna(subset=["transaction_id", "net_amount"])
    # Supply defaults for optional values.
    .fillna(
        {
            "city": "Unknown",
            "tax_amount": 0.0,
        }
    )
    # Apply the expected numeric types.
    .withColumn(
        "net_amount",
        F.col("net_amount").cast("double"),
    )
    .withColumn(
        "tax_amount",
        F.col("tax_amount").cast("double"),
    )
    # Keep one row for each transaction.
    .dropDuplicates(["transaction_id"])
)

4. Ukujoyina amasethi edatha ku-PySpark ngaphandle kokulahleka

Uma uke wasebenza ngezinqolobane zolwazi ngaphambilini, kungenzeka ukuthi ubhale izitatimende ze-SQL ezihlanganisa amathebula amabili noma ngaphezulu ndawonye. Ukujoyina ku-Spark kusebenza ngendlela efanayo, kodwa kuma-Dataframes.

Kuyini ukujoyina?

Uma umqondo wokujoyina umusha kuwe, ayindlela yokufanisa imigqa esuka ku-DataFrame eyodwa enemigqa ehlobene esuka kwenye i-DataFrame. Ngamanye amazwi, iphendula umbuzo othi:

“Imiphi imigqa kule DataFrame ehambisana nemigqa kuleyo DataFrame?”

Lowo umbono oyinhloko ngemuva kwakho konke ukujoyina ku-PySpark. Uma leyo ngxenye isicacile, i-syntax nezinhlobo zokujoyina ziba lula kakhulu ukuqonda.

Uma unamaDathaframes amabili afana nala:

sales_data.csv

transaction_id, customer_name, net_amount, tax_amount
101, Alice,    250.50, 25.05
102, Bob,      120.00, 6.00

amakhasimende.csv

customer_name, city, loyalty_level
Alice, New York, Gold
Bob,   London,   Silver

Ungabajoyina enkambini yabo evamile yegama lekhasimende kanje:

df_sales = spark.read.csv("sales_data.csv", header=True)
df_customers = spark.read.csv("customers.csv", header=True)
df_joined = df_sales.join(df_customers, on="customer_name", how="inner")
df_joined.show()


# Output
+-------------+--------------+----------+----------+--------+-------------+
|customer_name|transaction_id|net_amount|tax_amount|city    |loyalty_level|
+-------------+--------------+----------+----------+--------+-------------+
|Alice        |101           |250.50    |25.05     |New York|Gold         |
|Bob          |102           |120.00    |6.00      |London  |Silver       |
+-------------+--------------+----------+----------+--------+-------------+

Yikuphi ukujoyina okufanele abaqalayo bakusebenzise?

Kunezinhlobo eziningana ezahlukene zokujoyina ezitholakala e-Spark. Ku-99% wezimo zokusebenzisa zabaqalayo, uzosebenzisa okukodwa kwalokhu okulandelayo:

• ingaphakathi — khombisa kuphela imigqa efanayo
• left — khombisa yonke into kuthebula elingakwesokunxele, kanye nokufanayo
• outer — khombisa yonke imigqa kuwo womabili amathebula

Futhi kulokhu, ukujoyina kwangaphakathi kuzoba kude nakude uhlobo oluvame kakhulu lokujoyina ozolisebenzisa emsebenzini wakho wansuku zonke.

Ungakhathazeki “ngokusakaza”, “sort-merge”, “shuffle-hash”, nanoma iliphi elinye isu lokujoyina elithuthukisiwe okwamanje. Njengoba ulwazi lwakho lwe-Spark lukhula, ungafunda ngalokhu ngesikhathi sokuphumula kwakho.

Khumbula nje:

Ukujoyina kubiza kakhulu ngokwekhompiyutha kunokusebenza kwekholomu okululasngakho zisebenzise lapho kudingekile – kodwa hhayi ngokunganaki.

5. Ukufunda nokubhala idatha ngaphandle “ngendlela ye-Spark”: I-Parquet

Iningi labaqalayo linamathela ku-CSV ngoba lijwayele. Kodwa i-CSV iyanensa, ayiqinile, futhi ayinakho usekelo lwezinhlobo zedatha, futhi empilweni yangempela, i-Parquet iyifomethi yedatha yomdabu ye-Spark. I-Parquet iyikholomu, ifomethi yedatha ecindezelwe efaneleka kahle ukuhlaziya idatha, ukubikwa kwedatha kanye nemithwalo enzima efundwayo.

Uma uSpark efunda isethi yedatha yeParquet:

• Ilayisha kuphela amakholomu owadinga ngempela.
• Iqonda lonke uhlobo lwedatha.
• Ilayisha ngokushesha kakhulu kune-CSV.

Ubhala okuqukethwe kwe-Dataframe kumafayela efomethi ye-Parquet kanje:

df_joined.write.mode("overwrite").parquet("output/enriched_sales")

Khona-ke ungayifunda emuva ngokushesha kanje,

df_fast = spark.read.parquet("output/enriched_sales")
df_fast.show()

NB. Ukusebenzisa i-Parquet kokufaka nokukhipha ifayela kuwumsebenzi owodwa olula “wokuthuthukisa” kunoma yimuphi oqalayo we-Spark.

6. Ukucabanga ku-PySpark workflows

Uma usuqonda ukuthi ifundwa kanjani idatha, uyihlanze, uyiguqule, uyijoyine, futhi uyibhale futhi, isinyathelo esilandelayo ukufunda ukuhlela lezo zenzo zibe wuhlelo lokusebenza olulula. Iphrojekthi ye-PySpark eqalayo ivamise ukulandela lokhu kulandelana:

Read data
  -> check and clean it
  -> add useful columns
  -> combine with other data
  -> write the result

Lokho kungase kuzwakale kusobala, kodwa kuwushintsho olubalulekile. Awusazami nje nge-DataFrame eyodwa ngesikhathi. Wakha inqubo ephindaphindwayo.

Gcina isigaba ngasinye silula

Umkhuba owusizo osaqalayo ukunikeza isigaba ngasinye sokuhamba komsebenzi inhloso ecacile. Ngokwesibonelo:

df_raw = spark.read.schema(schema).csv("sales_data.csv", header=True)

df_clean = df_raw.dropna(subset=["net_amount", "tax_amount"])

df_enriched = df_clean.withColumn(
    "gross_amount",
    F.col("net_amount") + F.col("tax_amount")
)

df_final = df_enriched.join(df_customers, on="customer_name", how="left")

df_final.write.mode("overwrite").parquet("output/final_dataset")

Lesi sitayela sine-verbose kakhudlwana kunokuhlanganisa yonke into ibe yinkulumo eyodwa ende, kodwa kulula kakhulu ukuyifunda lapho ufunda.

Igama ngalinye le-DataFrame likutshela ukuthi ukuphi ekuhambeni komsebenzi:

df_raw       -> the data as it arrived
df_clean     -> the data after basic cleaning
df_enriched  -> the data after adding new meaning
df_final     -> the dataset ready to save

Kungani lokhu kubalulekile

Uma kukhona okungahambi kahle, lesi sakhiwo senza ukulungisa iphutha kube lula kakhulu.

Ungahlola isigaba ngasinye ngokubheka idatha:

df_raw.show()
df_clean.show()
df_enriched.show()

Ungahlola ukubalwa kwemigqa:

df_raw.count()
df_clean.count()
df_final.count()

Lokhu kusiza ukuphendula imibuzo ewusizo efana nalena:

Did rows disappear unexpectedly during cleaning?
Did the join create more rows than expected?
Did a calculated column produce nulls?

Imodeli yengqondo elula: Okufakiwe → ukulungiselela → inhlanganisela → okukhiphayo kuzokuyisa kude ngokumangalisayo ohambweni lwakho lwe-PySpark.

7. Isingeniso esimnene se-Spark UI

I-Spark ine-UI yewebhu enhle evulekayo lapho wenza isenzo esifana naso .count() noma .write(). Uma umsebenzi wakho we-Spark usebenza endaweni, vakashela:

Kufanele ubone okuthile okufana nalokhu kubonisiwe.

Kubukeka kukukhulu kakhulu, kodwa awudingi ukuqonda wonke amathebhu. Kulesi sigaba, udinga kuphela ukwazi ukuthi i-UI ikhona nokuthi kungani ilusizo. Futhi iwusizo ngoba ikusiza ukuthi ubone ukuthi yimiphi imisebenzi ye-Spark esebenzile noma esebenzayo njengamanje.

Futhi, njengoba ulwazi lwakho ku-Spark lukhula, i-UI ingakusiza uqonde ukuthi kungani imisebenzi yehlulekile noma ithatha isikhathi eside ukusebenza kunokulindelekile. Kodwa lokho kuza kamuva kakhulu. Okwamanje, phatha i-Spark UI njengedeshibhodi emotweni yakho — awudingi ukuqonda injini ukuze uqaphele uma okuthile kubukeka kuxakile.

Isifingqo: Manje usulungele iphrojekthi yakho yokuqala yangempela ye-PySpark

Kuleli qophelo, usudlulele kokuthi “Ngingasebenzisa i-Spark” ungene kokuthi “Ngingakha ipayipi elihlanzekile, elilula le-Spark.”

Manje uyazi ukuthi ungakwenza kanjani:

• funda idatha ngokuphepha,
• hlanza futhi uyilungiselele,
• cebisa ngamakholomu amasha,
• hlanganisa idathasethi eminingi,
• gcina umphumela ngempumelelo,
• futhi ubuke uSpark ngokwanele ukuze uhlale uqiniseka.

Akukho lutho kulesi sihloko oludinga iqoqo. Akukho okwakudinga ukushuna okuthuthukisiwe. Yile mpela mangaki amaphrojekthi we-PySpark wangempela aqalayo.

Uma usunolwazi oluthe xaxa, ungase ufune ukwakhela phezu kolwazi lwakho ngokucwaninga ezinye zalezi zihloko.

• ukufunda izinhlelo zokwenza
• ukuqonda ukushova
• ukuphatha ukwahlukanisa
• ezinye izinhlobo zokujoyina
• ukushuna ukusebenza okulula

Lezi ngezinye zezihloko engithemba ukuzokhuluma ngazo esihlokweni esizayo, kodwa okwamanje, usuphumelele ingqopha-mlando elandelayo, futhi ungakha okuthile okuzuzisayo nokuwusizo nge-PySpark.

---

BTW, nasi leso sixhumanisi esiya esihlokweni sokuqala kulolu chungechunge,
I-PySpark yabaqalayo: Ukuqonda Okuyisisekeloengikhulume ngayo ekuqaleni.