เพิ่มเติมเกี่ยวกับ Spark DataFrames

PySpark เบื้องต้น

Benjamin Schmidt

Data Engineer

การสร้าง DataFrames จากแหล่งข้อมูลต่างๆ

CSV Files: รูปแบบทั่วไปสำหรับข้อมูลที่มีโครงสร้างและตัวคั่น
JSON Files: รูปแบบข้อมูลกึ่งโครงสร้างแบบลำดับชั้น
Parquet Files: ปรับแต่งสำหรับการจัดเก็บและการคิวรี นิยมใช้ใน data engineering

ตัวอย่าง:
```
spark.read.csv("path/to/file.csv")
```
ตัวอย่าง:
```
spark.read.json("path/to/file.json")
```

ตัวอย่าง:

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

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

การอนุมาน schema และการกำหนด schema แบบ manual

Spark สามารถอนุมาน schema จากข้อมูลได้โดยใช้ inferSchema=True
กำหนด schema เองเพื่อควบคุมได้มากขึ้น — เหมาะสำหรับโครงสร้างข้อมูลที่คงที่

Schema ในระดับขนาดใหญ่

ชนิดข้อมูลใน PySpark DataFrames

IntegerType: จำนวนเต็ม
- เช่น 1, 3478, -1890456
LongType: จำนวนเต็มขนาดใหญ่
- เช่น จำนวนเต็มแบบ signed ขนาด 8 ไบต์ 922334775806
FloatType และ DoubleType: จำนวนทศนิยม
- เช่น 3.14159
StringType: ใช้สำหรับข้อความหรือข้อมูลชนิด string
- เช่น "This is an example of a string."
...

ไวยากรณ์ DataTypes สำหรับ 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)

การดำเนินการกับ DataFrame — การเลือกและกรองข้อมูล

ใช้ .select() เพื่อเลือกคอลัมน์ที่ต้องการ
ใช้ .filter() หรือ .where() เพื่อกรองแถวตามเงื่อนไข
ใช้ .sort() เพื่อเรียงลำดับตามกลุ่มคอลัมน์

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

การเรียงลำดับและการลบค่าที่หายไป

เรียงลำดับข้อมูลด้วย .sort() หรือ .orderBy()
ใช้ na.drop() เพื่อลบแถวที่มีค่า null

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

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

Cheatsheet

spark.read_json(): โหลดข้อมูลจาก JSON
spark.read.schema(): กำหนด schema อย่างชัดเจน
.na.drop(): ลบแถวที่มีค่าว่าง
.select(), .filter(), .sort(), .orderBy(): ฟังก์ชันพื้นฐานสำหรับการจัดการข้อมูล

มาฝึกกันเถอะ!

PySpark เบื้องต้น