<jupyter_start><jupyter_text>Spark High-Level API![img](https://static.packt-cdn.com/products/9781785888359/graphics/8bffbd94-04f7-46e3-a1e9-0d6046d2dcab.png)Source: https://static.packt-cdn.com/products/9781785888359/graphics/8bffbd94-04f7-46e3-a1e9-0d6046d2dcab.png Overview of Spark SQLCreating DataFrames- [ ] RDD    - [ ] createDataFrame()- [ ] Text file    - [ ] read.text()- [ ] JSON file    - [ ] read.json()    - [ ] read,json(RDD)- [ ] Parquet file    - [ ] read.parquet()- [ ] Table in a relational database- [ ] Temporary table in SparkDataFrame to RDD- [ ] rdd()Schemas- [ ] Inferring schemas    - [ ] Why it is not optimal practice- [ ] Specifying schemas    - [ ] Using StructType and StructField    - [ ] Using DDL string (schema = “author STRING, title STRING, pages INT”)- [ ] Metadata    - [ ] printSchema()    - [ ] columns()    - [ ] dtypes()- [ ] Actions    - [ ] show()- [ ] Transforms    - [ ] select() and alias()    - [ ] drop()    - [ ] filter() / where()    - [ ] distinct()    - [ ] dropDuplicates()    - [ ] sample    - [ ] sampleBy()    - [ ] limit()    - [ ] orderBy() / sort()    - [ ] groupBy()Operations that return an RDD     - [ ] rdd.map()    - [ ] rdd.flatMap()pyspark.sql.functions module- [ ] String functions- [ ] Math functions- [ ] Statistics functions- [ ] Date functions- [ ] Hashing functions- [ ] Algorithms (sounded, levenstein)- [ ] Windowing functionsUser defined functions- [ ] udf()- [ ] pandas_udf()Multiple DataFrames- [ ] join(other, on, how)- [ ] union(), unionAll()- [ ] intersect()- [ ] subtract()Persistence- [ ] cache()- [ ] persist(:- [ ] unpersist()- [ ] cacheTable()- [ ] clearCache()- [ ] repartition()- [ ] coalesce()Output- [ ] write.csv()- [ ] write.parquet()- [ ] write.json()Spark SQL- [ ] df.createOrReplaceTempView- [ ] sql()- [ ] table()  Implementation notesSpark 2.4 is desinged to run on Java 8 (or version 1.8). If you have veriosns of Java greater than Java 8, also install Java 8 and include this export```bashexport JAVA_HOME=$(/usr/libexec/java_home -v 1.8)```If using the latest `pyarrow` (`pip install -U pyarrow`), you will need to set this variable```bashARROW_PRE_0_15_IPC_FORMAT=1```in the file```bashSPARK_HOME/conf/spark-env.sh```<jupyter_code>from pyspark.sql import SparkSession
import pyspark.sql.functions as F
import pyspark.sql.types as T
spark = (
    SparkSession.builder 
    .master("local") 
    .appName("BIOS-823") 
    .config("spark.executor.cores", 4) 
    .getOrCreate()    
)
spark.version
spark.conf.get('spark.executor.cores')<jupyter_output><empty_output><jupyter_text>Create a Spark DataFrame<jupyter_code>df = spark.range(3)
df.show(3)
%%file data/test.csv
number,letter
0,a
1,c
2,b
3,a
4,b
5,c
6,a
7,a
8,a
9,b
10,b
11,c
12,c
13,b
14,b<jupyter_output><empty_output><jupyter_text>Implicit schema<jupyter_code>df = (
    spark.read.
    format('csv').
    option('header', 'true').
    option('inferSchema', 'true').
    load('data/test.csv')
)
df.show(3)
df.printSchema()<jupyter_output><empty_output><jupyter_text>Explicit schemaFor production use, you should provide an explicit schema to reduce risk of error.<jupyter_code>schema = T.StructType([
    T.StructField("number", T.DoubleType()),
    T.StructField("letter", T.StringType()),
])
df = (
    spark.read.
    format('csv').
    option('header', 'true').
    schema(schema).
    load('data/test.csv')
)
df.show(3)
df.printSchema()<jupyter_output><empty_output><jupyter_text>Alternative way to specify schemaYou can use SQL DDL syntax to specify a schema as well.<jupyter_code>schema = 'number DOUBLE, letter STRING'
df_altschema = (
    spark.read.
    format('csv').
    option('header', 'true').
    schema(schema=schema).
    load('data/test.csv')
)
df_altschema.take(3)
df_altschema.printSchema()<jupyter_output><empty_output><jupyter_text>Persist<jupyter_code>df.cache()<jupyter_output><empty_output><jupyter_text>Data manipulation  Select<jupyter_code>df.select('number').show(3)
from pyspark.sql.functions import col, expr
df.select(col('number').alias('index')).show(3)
df.select(expr('number as x')).show(3)
df.withColumnRenamed('number', 'x').show(3)<jupyter_output><empty_output><jupyter_text>Filter<jupyter_code>df.filter('number % 2 == 0').show(3)<jupyter_output><empty_output><jupyter_text>Sort<jupyter_code>df.sort(df.number.desc()).show(3)
df.orderBy(df.letter.desc()).show(3)<jupyter_output><empty_output><jupyter_text>Transform<jupyter_code>df.selectExpr('number*2 as x').show(3)
df.withColumn('x', expr('number*2')).show(3)<jupyter_output><empty_output><jupyter_text>Sumarize<jupyter_code>import pyspark.sql.functions as F
df.agg(F.min('number'),
       F.max('number'), 
       F.min('letter'), 
       F.max('letter')).show(3)<jupyter_output><empty_output><jupyter_text>Group by<jupyter_code>(
    df.groupby('letter').
    agg(F.mean('number'), F.stddev_samp('number')).show()
)<jupyter_output><empty_output><jupyter_text>Window functions<jupyter_code>from pyspark.sql.window import Window
ws = (
    Window.partitionBy('letter').
    orderBy(F.desc('number')).
    rowsBetween(Window.unboundedPreceding, Window.currentRow)
)
df.groupby('letter').agg(F.sum('number')).show()
df.show()
(
    df.select('letter', F.sum('number').
              over(ws).
              alias('rank')).show()
)<jupyter_output><empty_output><jupyter_text>SQL<jupyter_code>df.createOrReplaceTempView('df_table')
spark.sql('''SELECT * FROM df_table''').show(3)
spark.sql('''
SELECT letter, mean(number) AS mean, 
stddev_samp(number) AS sd from df_table
WHERE number % 2 = 0
GROUP BY letter
ORDER BY letter DESC
''').show()<jupyter_output><empty_output><jupyter_text>String operatons<jupyter_code>from pyspark.sql.functions import split, lower, explode
import pandas as pd
s = spark.createDataFrame(
    pd.DataFrame(
        dict(sents=('Thing 1 and Thing 2',
                    'The Quick Brown Fox'))))
s.show()
from pyspark.sql.functions import regexp_replace
s1 = (
    s.select(explode(split(lower(expr('sents')), ' '))).
    sort('col')
)
s1.show()
s1.groupBy('col').count().show()
s.createOrReplaceTempView('s_table')
spark.sql('''
SELECT regexp_replace(sents, 'T.*?g', 'FOO')
FROM s_table
''').show()<jupyter_output><empty_output><jupyter_text>Numeric operations<jupyter_code>from pyspark.sql.functions import log1p, randn
df.selectExpr('number', 'log1p(number)', 'letter').show(3)
(
    df.selectExpr('number', 'randn() as random').
    stat.corr('number', 'random')
)<jupyter_output><empty_output><jupyter_text>Date and time<jupyter_code>dt = (
    spark.range(3).
    withColumn('today', F.current_date()).
    withColumn('tomorrow', F.date_add('today', 1)).
    withColumn('time', F.current_timestamp())
)
dt.show()
dt.printSchema()<jupyter_output><empty_output><jupyter_text>Nulls<jupyter_code>%%file data/test_null.csv
number,letter
0,a
1,
2,b
3,a
4,b
5,
6,a
7,a
8,
9,b
10,b
11,c
12,
13,b
14,b
dn = (
    spark.read.
    format('csv').
    option('header', 'true').
    option('inferSchema', 'true').
    load('data/test_null.csv')
)
dn.printSchema()
dn.show()
dn.na.drop().show()
dn.na.fill('Missing').show()<jupyter_output><empty_output><jupyter_text>UDFTo avoid degrading performance, avoid using UDF if you can use the functions in `pyspark.sql.functions`. If you must use UDFs, prefer `pandas_udf` to `udf` where possible.<jupyter_code>from pyspark.sql.functions import udf, pandas_udf<jupyter_output><empty_output><jupyter_text>Standard Python UDF<jupyter_code>@udf('double')
def square(x):
    return x**2
df.select('number', square('number')).show(3)<jupyter_output><empty_output><jupyter_text>Pandas UDF<jupyter_code>@pandas_udf('double')
def scale(x):
    return (x - x.mean())/x.std()
df.select(scale('number')).show(3)<jupyter_output><empty_output><jupyter_text>Grouped agg<jupyter_code>@pandas_udf('double', F.PandasUDFType.GROUPED_AGG)
def gmean(x):
    return x.mean()
df.groupby('letter').agg(gmean('number')).show()<jupyter_output><empty_output><jupyter_text>Grouped map<jupyter_code>@pandas_udf(df.schema, F.PandasUDFType.GROUPED_MAP)
def gscale(pdf):
    return pdf.assign(
        number = (pdf.number - pdf.number.mean()) /
        pdf.number.std())
df.groupby('letter').apply(gscale).show()<jupyter_output><empty_output><jupyter_text>Joins<jupyter_code>names = '<NAME>'.split()
courses = '821 823 821 824 823'.split()
pdf1 = pd.DataFrame(dict(name=names, course=courses))
pdf1
course_id = '821 822 823 824 825'.split()
course_names = 'Unix Python R Spark GLM'.split()
pdf2 = pd.DataFrame(dict(course_id=course_id, name=course_names))
pdf2
df1 = spark.createDataFrame(pdf1)
df2 = spark.createDataFrame(pdf2)
df1.join(df2, df1.course == df2.course_id, how='inner').show()
df1.join(df2, df1.course == df2.course_id, how='right').show()<jupyter_output><empty_output><jupyter_text>DataFrame conversions<jupyter_code>sc = spark.sparkContext
rdd = sc.parallelize([('ann', 23), ('bob', 34)])
df = spark.createDataFrame(rdd, schema='name STRING, age INT')
df.show()
df.rdd.map(lambda x: (x[0], x[1]**2)).collect()
df.rdd.mapValues(lambda x: x**2).collect()
df.toPandas()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.17335385291854458, 'fraction_numerical': 0.022644827965263275, 'mean_word_length': 5.08561872909699, 'pattern_counts': {'":': 0, '<': 102, '<?xml version=': 0, '>': 102, 'https://': 2, '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': '25', 'dir': 'jupyter-structured-clean-dedup', 'id': '156864', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 3543, 'n_tokens_neox': 3399, 'n_words': 580, 'path': 'notebooks/16_Spark_High_Level_API.ipynb', 'repo_name': 'yuan-feng1/bios-823-2019', 'size': '9112'}