Więcej o Spark DataFrames

Wprowadzenie do PySpark

Benjamin Schmidt

Data Engineer

Tworzenie DataFrame z różnych źródeł danych

Pliki CSV: Popularne dla danych strukturalnych, ograniczonych separatorem
Pliki JSON: Półstrukturalny, hierarchiczny format danych
Pliki Parquet: Zoptymalizowane pod kątem przechowywania i zapytań, często stosowane w inżynierii danych

Przykład:
```
spark.read.csv("path/to/file.csv")
```
Przykład:
```
spark.read.json("path/to/file.json")
```

Przykład:

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

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

Wnioskowanie i ręczna definicja schematu

Spark może wywnioskować schemat z danych przy użyciu inferSchema=True
Ręczna definicja schematu zapewnia lepszą kontrolę – przydatna dla stałych struktur danych

Schemat na dużą skalę

Typy danych w PySpark DataFrames

IntegerType: Liczby całkowite
- Np. 1, 3478, -1890456
LongType: Większe liczby całkowite
- Np. 8-bajtowe liczby ze znakiem, 922334775806
FloatType i DoubleType: Liczby zmiennoprzecinkowe dla wartości dziesiętnych
- Np. 3.14159
StringType: Dane tekstowe
- Np. "This is an example of a string."
...

Składnia typów danych w 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)

Operacje na DataFrame – wybór i filtrowanie

Użyj .select(), aby wybrać kolumny
Użyj .filter() lub .where(), aby filtrować wiersze
Użyj .sort(), aby posortować według kolumn

# 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()

Sortowanie i usuwanie brakujących wartości

Sortowanie za pomocą .sort() lub .orderBy()
Użyj na.drop(), aby usunąć wiersze z wartościami null

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

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

Ściągawka

spark.read_json(): Wczytuje dane z JSON
spark.read.schema(): Jawnie definiuje schemat
.na.drop(): Usuwa wiersze z brakującymi wartościami
.select(), .filter(), .sort(), .orderBy(): Podstawowe funkcje manipulacji danymi

Czas na ćwiczenia!

Wprowadzenie do PySpark