Агрегації в PySpark

Вступ до PySpark

Benjamin Schmidt

Data Engineer

Огляд SQL-агрегацій у PySpark

Звичні SQL-агрегації працюють через spark.sql()

# SQL-агрегація
spark.sql("""
  SELECT Department, SUM(Salary) AS Total_Salary, AVG(Salary) AS Average_Salary
  FROM employees
  GROUP BY Department
""").show()

Поєднання операцій DataFrame і SQL

# Відфільтруйте зарплати понад 3000
filtered_df = df.filter(df.Salary > 3000)

# Зареєструйте відфільтрований DataFrame як подання
filtered_df.createOrReplaceTempView("filtered_employees")


# Агрегація SQL на відфільтрованому поданні
spark.sql("""
    SELECT Department, COUNT(*) AS Employee_Count
    FROM filtered_employees
    GROUP BY Department
""").show()

Типи даних в агрегаціях

# Приклад зміни типу
data = [("HR", "3000"), ("IT", "4000"), ("Finance", "3500")]
columns = ["Department", "Salary"]
df = spark.createDataFrame(data, schema=columns)


# Перетворіть стовпець Salary на ціле число
df = df.withColumn("Salary", df["Salary"].cast("int"))

# Виконайте агрегацію
df.groupBy("Department").sum("Salary").show()

Агрегації з RDD

# Приклад агрегації з RDD
rdd = df.rdd.map(lambda row: (row["Department"], row["Salary"]))


rdd_aggregated = rdd.reduceByKey(lambda x, y: x + y)


print(rdd_aggregated.collect())

Найкращі практики для агрегацій у PySpark

Фільтруйте рано: зменшуйте обсяг даних перед агрегаціями
Стежте за типами: дані мають бути чистими й коректно типізованими
Уникайте операцій над усім набором: мінімізуйте, зокрема, groupBy()
Обирайте інтерфейс: надавайте перевагу DataFrame завдяки оптимізаціям
Моніторте продуктивність: використовуйте explain() для плану виконання та подальшої оптимізації

Основні висновки

SQL-агрегації в PySpark: функції SUM() і AVERAGE() для підсумків даних
DataFrame і SQL: комбінуйте підходи для гнучкого опрацювання
Типи даних: усувайте невідповідності типів під час агрегацій
RDD проти DataFrame: розумійте компроміси й обирайте доречний інструмент

Давайте потренуємось!

Вступ до PySpark