Mai multe despre Spark DataFrames

Introducere în PySpark

Benjamin Schmidt

Data Engineer

Crearea DataFrame-urilor din diverse surse de date

Fișiere CSV: Uzuale pentru date structurate, delimitate
Fișiere JSON: Format semi-structurat, ierarhic
Fișiere Parquet: Optimizate pentru stocare și interogare, frecvente în ingineria datelor

Exemplu:
```
spark.read.csv("path/to/file.csv")
```
Exemplu:
```
spark.read.json("path/to/file.json")
```

Exemplu:

spark.read.parquet("path/to/file.parquet")

¹ https://spark.apache.org/docs/latest/api/python/reference/pyspark.pandas/api/pyspark.pandas.read_csv

Inferența schemei și definirea manuală a schemei

Spark poate deduce schemele din date cu inferSchema=True
Definiți manual schema pentru un control mai bun - util pentru structuri fixe de date

Schemă la scară

Tipuri de date în PySpark DataFrames

IntegerType: Numere întregi
- Ex.: 1, 3478, -1890456
LongType: Numere întregi mai mari
- Ex.: numere semnate pe 8 octeți, 922334775806
FloatType și DoubleType: Numere în virgulă mobilă pentru valori zecimale
- Ex.: 3.14159
StringType: Utilizat pentru date text sau șiruri de caractere
- Ex.: "This is an example of a string."
...

Sintaxa tipurilor de date pentru PySpark DataFrames

# Import the necessary types as classes
from pyspark.sql.types import (StructType,
                            StructField, IntegerType,
                            StringType, ArrayType)

# Construct the schema
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("scores", ArrayType(IntegerType()), True)
])

# Set the schema
df = spark.createDataFrame(data, schema=schema)

Operații pe DataFrame - selecție și filtrare

Utilizați .select() pentru a alege coloane specifice
Utilizați .filter() sau .where() pentru a filtra rânduri după condiții
Utilizați .sort() pentru a ordona după coloane

# Select and show only the name and age columns
df.select("name", "age").show()

# Filter on age > 30
df.filter(df["age"] > 30).show()

# Use Where to filter match a specific value
df.where(df["age"] == 30).show()

# Use Sort to sort on age
df.sort("age", ascending=False).show()

Sortare și eliminarea valorilor lipsă

Ordonați datele cu .sort() sau .orderBy()
Utilizați na.drop() pentru a elimina rândurile cu valori nule

# Sort using the age column
df.sort("age", ascending=False).show()

# Drop missing values
df.na.drop().show()

Cheatsheet

spark.read_json(): Încărcați date din JSON
spark.read.schema(): Definiți scheme explicit
.na.drop(): Eliminați rândurile cu valori lipsă
.select(), .filter(), .sort(), .orderBy(): Funcții de bază pentru manipularea datelor

Să exersăm!

Introducere în PySpark