<jupyter_start><jupyter_text>Spark SQLUsing Spark SQL and DataFrames results in performance that is comparable to native compiled Scala code. This is not true in general if you use the low-level RDD API.![img](https://miro.medium.com/max/1524/1*Kolwjg356xdqPv-8JmQGRQ.png)Source: https://miro.medium.com/max/1524/1*Kolwjg356xdqPv-8JmQGRQ.pngThe improvement in efficiency is largely due to the Catalyst optimizer, which applies rule-based and cost-based optimization. Optimization occurs in 4 phases:1. Analysis    - Generate an abstract syntax tree    - Resolve names using internal catalog2. Logical optimization    - Construct plans using rule-based optimization    - Assign costs to each plan using cost-based optimization3. Physical planning    - Generate a physical plan using operations available in Spark execution engine4. Code generation    - Generate Java bytecode to run on each machine    - Generates compact RDD code for final execution![imng](https://miro.medium.com/max/1400/1*_DdwvGk23tB3p1aLQB3Q4A.png)Source: https://miro.medium.com/max/1400/1*_DdwvGk23tB3p1aLQB3Q4A.pngjpg<jupyter_code>from pyspark.sql import SparkSession
spark = (
    SparkSession.builder 
    .master("local") 
    .appName("BIOS-823") 
    .config("spark.executor.cores", 4) 
    .getOrCreate()    
)
import pandas as pd
df = spark.createDataFrame(pd.DataFrame(dict(a=range(6), b=list('aabbcc'))))
df.show()<jupyter_output>+---+---+
|  a|  b|
+---+---+
|  0|  a|
|  1|  a|
|  2|  b|
|  3|  b|
|  4|  c|
|  5|  c|
+---+---+<jupyter_text>Read the DataFrame into a temporary view for SQL queries.<jupyter_code>df.createOrReplaceTempView('df_table')<jupyter_output><empty_output><jupyter_text>If you want to create a permanent table, use this ```pythondf.write.saveAsTable('df_table')```<jupyter_code>spark.catalog.listDatabases()
spark.catalog.listTables()<jupyter_output><empty_output><jupyter_text>To convert a table back inot a DataFrame<jupyter_code>spark.sql('SELECT * FROM df_table').show()
spark.table('df_table').show()<jupyter_output>+---+---+
|  a|  b|
+---+---+
|  0|  a|
|  1|  a|
|  2|  b|
|  3|  b|
|  4|  c|
|  5|  c|
+---+---+<jupyter_text>Standard SQL QueriesAll Spark SQL queries return a DataFrame.<jupyter_code>spark.sql('''
SELECT a, b FROM df_table
''').show()
spark.sql('''
SELECT * 
FROM df_table
WHERE b='b'
''').show()
spark.sql('''
SELECT b, mean(a) AS avg, std(a) AS std
FROM df_table
GROUP BY b
ORDER BY b
''').show()<jupyter_output>+---+---+------------------+
|  b|avg|               std|
+---+---+------------------+
|  a|0.5|0.7071067811865476|
|  b|2.5|0.7071067811865476|
|  c|4.5|0.7071067811865476|
+---+---+------------------+<jupyter_text>Functions<jupyter_code>spark.sql('''
SHOW FUNCTIONS "c*"
''').show(50)
from pyspark.sql import functions
sorted([f for f in dir(functions) if f.startswith('c')])<jupyter_output><empty_output><jupyter_text>Complex Types  Structs<jupyter_code>df1 = spark.sql('''
SELECT a, b, (a, b) AS struct FROM df_table
''')
df1.show()
df1.createOrReplaceTempView('df1_table')
spark.sql('''
SELECT struct.a, struct.b FROM df1_table
''').show()<jupyter_output>+---+---+
|  a|  b|
+---+---+
|  0|  a|
|  1|  a|
|  2|  b|
|  3|  b|
|  4|  c|
|  5|  c|
+---+---+<jupyter_text>Lists<jupyter_code>df2 = spark.sql('''
SELECT b, collect_list(a) as list
FROM df_table
GROUP BY b
ORDER BY b
''')
df2.createOrReplaceTempView('df2_table')
df2.show()
spark.sql('''
SELECT b, explode(list) as a
FROM df2_table
''').show()<jupyter_output>+---+---+
|  b|  a|
+---+---+
|  a|  0|
|  a|  1|
|  b|  2|
|  b|  3|
|  c|  4|
|  c|  5|
+---+---+<jupyter_text>Maps<jupyter_code>from pyspark.sql.functions import create_map
from itertools import chain
df3 = df.select('a', 'b', create_map('b', 'a').alias('map'))
df3.createOrReplaceTempView('df3_table')
df3.show()
spark.sql('''
SELECT a, b, map['c'] FROM df3_table
''').na.drop().show()
df3.select('a', 'b', 'map.c').filter('c is not null').show()<jupyter_output>+---+---+---+
|  a|  b|  c|
+---+---+---+
|  4|  c|  4|
|  5|  c|  5|
+---+---+---+
{'fraction_non_alphanumeric': 0.1991546494281452, 'fraction_numerical': 0.03804077573346594, 'mean_word_length': 3.7609467455621304, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '648495', 'license': 'MIT', 'n_tokens_mistral': 1702, 'n_tokens_neox': 1625, 'n_words': 360, 'path': 'notebooks/C05_Spark_SQL.ipynb', 'repo_name': 'cliburn/bios-823-2020', 'size': '4022'}