Работа с DataFrame в PySpark SQL
Основы Big Data с PySpark
Upendra Devisetty
Science Analyst, CyVerse
DataFrame API и SQL-запросы
В PySpark взаимодействие с SparkSQL возможно через DataFrame API и SQL-запросы
DataFrame API предоставляет предметно-ориентированный язык (DSL) для работы с данными
Трансформации и действия с DataFrame удобно выстраивать программно
SQL-запросы лаконичны, понятны и переносимы
Операции с DataFrame также можно выполнять с помощью SQL-запросов
Выполнение SQL-запросов
Метод
sql()объекта SparkSession выполняет SQL-запросsql()принимает SQL-выражение в виде аргумента и возвращает результат в виде DataFrame
df.createOrReplaceTempView("table1")
df2 = spark.sql("SELECT field1, field2 FROM table1")
df2.collect()
[Row(f1=1, f2='row1'), Row(f1=2, f2='row2'), Row(f1=3, f2='row3')]
SQL-запрос для извлечения данных
test_df.createOrReplaceTempView("test_table")
query = '''SELECT Product_ID FROM test_table'''
test_product_df = spark.sql(query)
test_product_df.show(5)
+----------+
|Product_ID|
+----------+
| P00069042|
| P00248942|
| P00087842|
| P00085442|
| P00285442|
+----------+
Агрегация и группировка данных с помощью SQL-запросов
test_df.createOrReplaceTempView("test_table")
query = '''SELECT Age, max(Purchase) FROM test_table GROUP BY Age'''
spark.sql(query).show(5)
+-----+-------------+
| Age|max(Purchase)|
+-----+-------------+
|18-25| 23958|
|26-35| 23961|
| 0-17| 23955|
|46-50| 23960|
|51-55| 23960|
+-----+-------------+
only showing top 5 rows
Фильтрация столбцов с помощью SQL-запросов
test_df.createOrReplaceTempView("test_table")
query = '''SELECT Age, Purchase, Gender FROM test_table WHERE Purchase > 20000 AND Gender == "F"'''
spark.sql(query).show(5)
+-----+--------+------+
| Age|Purchase|Gender|
+-----+--------+------+
|36-45| 23792| F|
|26-35| 21002| F|
|26-35| 23595| F|
|26-35| 23341| F|
|46-50| 20771| F|
+-----+--------+------+
only showing top 5 rows
Давайте потренируемся!
Основы Big Data с PySpark
