<jupyter_start><jupyter_text>PySpark Demo Notebook Demo1. Run PostgreSQL Script2. Load PostgreSQL Data3. Create New Record4. Write New Record to PostgreSQL Table5. Load CSV Data File6. Write Data to PostgreSQL7. Analyze Data with Spark SQL8. Graph Data with BokehJS9. Read and Write Data to Parquet Format_Prepared by: [<NAME>](https://twitter.com/GaryStafford)   Associated article: https://wp.me/p1RD28-61V_  Run PostgreSQL ScriptRun the PostgreSQL sql script<jupyter_code>! pip install psycopg2-binary --upgrade --quiet
%run -i '03_load_sql.py'
from pyspark.sql import SparkSession
from pyspark.sql.functions import to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
spark = SparkSession \
    .builder \
    .appName('pyspark_demo_app') \
    .config('spark.driver.extraClassPath',
            'postgresql-42.2.10.jar') \
    .master("local[*]") \
    .getOrCreate()<jupyter_output><empty_output><jupyter_text>Load PostgreSQL DataLoad the PostgreSQL 'bakery_basket' table's contents into a DataFrame<jupyter_code>properties = {
    'driver': 'org.postgresql.Driver',
    'url': 'jdbc:postgresql://postgres:5432/demo',
    'user': 'postgres',
    'password': '<PASSWORD>',
    'dbtable': 'bakery_basket',
}

df1 = spark.read \
    .format('jdbc') \
    .option('driver', properties['driver']) \
    .option('url', properties['url']) \
    .option('user', properties['user']) \
    .option('password', properties['password']) \
    .option('dbtable', properties['dbtable']) \
    .load()
%%time
df1.show(10)
df1.count()<jupyter_output>+---+----------+--------+-----------+-------------+
| id|      date|    time|transaction|         item|
+---+----------+--------+-----------+-------------+
|  1|2016-10-30|09:58:11|          1|        Bread|
|  2|2016-10-30|10:05:34|          2| Scandinavian|
|  3|2016-10-30|10:07:57|          3|Hot chocolate|
+---+----------+--------+-----------+-------------+

CPU times: user 0 ns, sys: 0 ns, total: 0 ns
Wall time: 3.98 s<jupyter_text>Create New RecordCreate a new bakery record and load into a DataFrame<jupyter_code>data = [('2016-10-30', '10:13:27', 2, 'Pastry')]

bakery_schema = StructType([
    StructField('date', StringType(), True),
    StructField('time', StringType(), True),
    StructField('transaction', IntegerType(), True),
    StructField('item', StringType(), True)
])

df2 = spark.createDataFrame(data, bakery_schema)
df2.show()
df2.count()<jupyter_output>+----------+--------+-----------+------+
|      date|    time|transaction|  item|
+----------+--------+-----------+------+
|2016-10-30|10:13:27|          2|Pastry|
+----------+--------+-----------+------+<jupyter_text>Write New Record to PostgreSQL TableAppend the contents of the DataFrame to the PostgreSQL 'bakery_basket' table<jupyter_code>df2.write \
    .format('jdbc') \
    .option('driver', properties['driver']) \
    .option('url', properties['url']) \
    .option('user', properties['user']) \
    .option('password', properties['password']) \
    .option('dbtable', properties['dbtable']) \
    .mode('append') \
    .save()
df1.show(10)
df1.count()<jupyter_output>+---+----------+--------+-----------+-------------+
| id|      date|    time|transaction|         item|
+---+----------+--------+-----------+-------------+
|  1|2016-10-30|09:58:11|          1|        Bread|
|  2|2016-10-30|10:05:34|          2| Scandinavian|
|  3|2016-10-30|10:07:57|          3|Hot chocolate|
|  1|2016-10-30|10:13:27|          2|       Pastry|
+---+----------+--------+-----------+-------------+<jupyter_text>Load CSV File DataLoad the Kaggle dataset from the CSV file, containing ~21K records, into a DataFrame<jupyter_code>! ls -lh *.csv
df3 = spark.read \
    .format("csv") \
    .option("header", "true") \
    .load("BreadBasket_DMS.csv", schema=bakery_schema)
df3.show(10)
df3.count()<jupyter_output>+----------+--------+-----------+-------------+
|      date|    time|transaction|         item|
+----------+--------+-----------+-------------+
|2016-10-30|09:58:11|          1|        Bread|
|2016-10-30|10:05:34|          2| Scandinavian|
|2016-10-30|10:05:34|          2| Scandinavian|
|2016-10-30|10:07:57|          3|Hot chocolate|
|2016-10-30|10:07:57|          3|          Jam|
|2016-10-30|10:07:57|          3|      Cookies|
|2016-10-30|10:08:41|          4|       Muffin|
|2016-10-30|10:13:03|          5|       Coffee|
|2016-10-30|10:13:03|          5|       Pastry|
|2016-10-30|10:13:03|          5|        Bread|
+----------+--------+-----------+-------------+
only showing top 10 rows<jupyter_text>Write Data to PostgreSQLAppend the contents of the DataFrame to the PostgreSQL 'bakery_basket' table<jupyter_code>df3.write \
    .format('jdbc') \
    .option('driver', properties['driver']) \
    .option('url', properties['url']) \
    .option('user', properties['user']) \
    .option('password', properties['password']) \
    .option('dbtable', properties['dbtable']) \
    .mode('append') \
    .save()
df1.show(10)
df1.count()<jupyter_output>+---+----------+--------+-----------+-------------+
| id|      date|    time|transaction|         item|
+---+----------+--------+-----------+-------------+
|  1|2016-10-30|09:58:11|          1|        Bread|
|  2|2016-10-30|10:05:34|          2| Scandinavian|
|  3|2016-10-30|10:07:57|          3|Hot chocolate|
|  1|2016-10-30|10:13:27|          2|       Pastry|
|  2|2016-10-30|09:58:11|          1|        Bread|
|  3|2016-10-30|10:05:34|          2| Scandinavian|
|  4|2016-10-30|10:05:34|          2| Scandinavian|
|  5|2016-10-30|10:07:57|          3|Hot chocolate|
|  6|2016-10-30|10:07:57|          3|          Jam|
|  7|2016-10-30|10:07:57|          3|      Cookies|
+---+----------+--------+-----------+-------------+
only showing top 10 rows<jupyter_text>Analyze Data with Spark SQLAnalyze the DataFrame's bakery data using Spark SQL<jupyter_code>df1.createOrReplaceTempView("bakery_table")
df4 = spark.sql("SELECT * FROM bakery_table " +
                "ORDER BY transaction, date, time")
df4.show(10)
df4.count()
df5 = spark.sql("SELECT COUNT(DISTINCT item) AS item_count FROM bakery_table")
df5.show()

df5 = spark.sql("SELECT item, count(*) as count " +
                "FROM bakery_table " +
                "WHERE item NOT LIKE 'NONE' " +
                "GROUP BY item ORDER BY count DESC " +
                "LIMIT 10")
df5.show()<jupyter_output>+----------+
|item_count|
+----------+
|        95|
+----------+

+-------------+-----+
|         item|count|
+-------------+-----+
|       Coffee| 5471|
|        Bread| 3326|
|          Tea| 1435|
|         Cake| 1025|
|       Pastry|  857|
|     Sandwich|  771|
|    Medialuna|  616|
|Hot chocolate|  591|
|      Cookies|  540|
|      Brownie|  379|
+-------------+-----+<jupyter_text>Graph Data with BokehJSCreate a vertical bar chart displaying DataFrame data<jupyter_code>from bokeh.io import output_notebook, show
from bokeh.plotting import figure
from bokeh.models import ColumnDataSource
from bokeh.transform import factor_cmap
from bokeh.palettes import Paired12

output_notebook()

source = ColumnDataSource(data=df5.toPandas())

tooltips = [('item', '@item'), ('count', '@{count}{,}')]

items = source.data['item'].tolist()
color_map = factor_cmap(field_name='item', palette=Paired12, factors=items)
plot = figure(x_range=items, plot_width=750, plot_height=375, min_border=0, tooltips=tooltips)
plot.vbar(x='item', bottom=0, top='count', source=source, width=0.9, fill_color=color_map)
plot.title.text = 'Top 10 Bakery Items'
plot.xaxis.axis_label = 'Bakery Items'
plot.yaxis.axis_label = 'Total Items Sold'

show(plot)
df6 = spark.sql("SELECT CONCAT(date,' ',time) as timestamp, transaction, item " +
                "FROM bakery_table " +
                "WHERE item NOT LIKE 'NONE'" +
                "ORDER BY transaction"
               )
df6.show(10)
df6.count()
df7 = df6.withColumn('timestamp', to_timestamp(df6.timestamp, 'yyyy-MM-dd HH:mm:ss'))
df7.printSchema()
df7.show(10)
df7.count()
df7.createOrReplaceTempView("bakery_table")
df8 = spark.sql("SELECT DISTINCT * " +
                "FROM bakery_table " +
                "WHERE item NOT LIKE 'NONE'" +
                "ORDER BY transaction DESC"
                )
df8.show(10)
df8.count()<jupyter_output>+-------------------+-----------+--------------+
|          timestamp|transaction|          item|
+-------------------+-----------+--------------+
|2017-04-09 15:04:24|       9684|     Smoothies|
|2017-04-09 14:57:06|       9683|        Pastry|
|2017-04-09 14:57:06|       9683|        Coffee|
|2017-04-09 14:32:58|       9682|           Tea|
|2017-04-09 14:32:58|       9682|  Tacos/Fajita|
|2017-04-09 14:32:58|       9682|        Muffin|
|2017-04-09 14:32:58|       9682|        Coffee|
|2017-04-09 14:30:09|       9681|Spanish Brunch|
|2017-04-09 14:30:09|       9681|      Truffles|
|2017-04-09 14:30:09|       9681|           Tea|
+-------------------+-----------+--------------+
only showing top 10 rows<jupyter_text>Read and Write Data to Parquet FormatRead and write DataFrame data to Parquet format files<jupyter_code>df8.write.parquet('output/bakery_parquet', mode='overwrite')
! ls -lh output/
df9 = spark.read.parquet('output/bakery_parquet')
df9.show(10)
df9.count()<jupyter_output>+-------------------+-----------+---------+
|          timestamp|transaction|     item|
+-------------------+-----------+---------+
|2017-02-15 14:54:25|       6620|     Cake|
|2017-02-15 14:41:27|       6619|    Bread|
|2017-02-15 14:40:41|       6618|   Coffee|
|2017-02-15 14:40:41|       6618|    Bread|
|2017-02-15 14:23:16|       6617| Baguette|
|2017-02-15 14:23:16|       6617|   Coffee|
|2017-02-15 14:23:16|       6617|    Salad|
|2017-02-15 14:23:16|       6617| Art Tray|
|2017-02-15 14:23:16|       6617|Alfajores|
|2017-02-15 14:16:26|       6616|    Bread|
+-------------------+-----------+---------+
only showing top 10 rows
{'fraction_non_alphanumeric': 0.24394184168012925, 'fraction_numerical': 0.0976373182552504, 'mean_word_length': 2.1069636135508154, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, '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': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '185863', 'license': 'MIT', 'n_tokens_mistral': 4005, 'n_tokens_neox': 3385, 'n_words': 680, 'path': 'work/04_pyspark_demo_notebook.ipynb', 'repo_name': 'Richaro10/pyspark-setup-demo', 'size': '9916'}