Interacțiunea cu DataFrames folosind PySpark SQL
Fundamentele Big Data cu PySpark
Upendra Devisetty
Science Analyst, CyVerse
API DataFrame vs. interogări SQL
În PySpark puteți interacționa cu SparkSQL prin API-ul DataFrame și interogări SQL
API-ul DataFrame oferă un limbaj specific domeniului (DSL) pentru date
Transformările și acțiunile pe DataFrame sunt mai ușor de construit programatic
Interogările SQL pot fi concise, ușor de înțeles și portabile
Operațiunile pe DataFrame pot fi realizate și prin interogări SQL
Executarea interogărilor SQL
Metoda
sql()din SparkSession execută o interogare SQLsql()primește o instrucțiune SQL ca argument și returnează rezultatul ca 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')]
Extragerea datelor cu interogări 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|
+----------+
Rezumarea și gruparea datelor cu interogări 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
Filtrarea coloanelor cu interogări 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
Timp de practică!
Fundamentele Big Data cu PySpark
