Interactive pySpark Session – JUPYTER.AI MINIBOOKS

Configuring and using SparkSession, SparkContext, DataFrameReader and DataStreamReader objects

Configure a SparkSession, SparkContext, DataFrameReader and DataStreamReader object. With minor tweaks — mostly to the conf object k/v pairs — the following initialization code can be use virtually anywhere. Assuming you’ve pip-installed pyspark, to start an ad-hoc interactive session, save the first code block to, say, ./pyspark_init.py, then run it as follows:

nmvega@fedora$ ptpython -i ./pyspark_init.py # Use python(1) if you don’t use ptpython.
# NOTE: For REPL sessions, your humble author prefers ptpython with vim(1) key bindings.

import sys, os
from pyspark.conf import SparkConf
from pyspark.sql import SparkSession, Catalog
from pyspark.sql import DataFrame, DataFrameStatFunctions, DataFrameNaFunctions
from pyspark.sql import functions as F # Spark Functions. USAGE: F.col(), F.countDistinct()
from pyspark.sql import types as T     # Spark Types.     USAGE: T.Integer()
from pyspark.sql.types import Row
  # ------------------------------------------
  # Note: Row() in .../pyspark/sql/types.py
  # isn't included in '__all__' list(), so
  # we must import it by name here.
  # ------------------------------------------

PROJECT_BASE_DIR        = '/home/nmvega/agile.ds.d/'
SPARK_WAREHOUSE_DIR     = PROJECT_BASE_DIR + 'stack.d/spark.d/spark-warehouse'
SPARK_CONF_DIR          = PROJECT_BASE_DIR + 'stack.d/spark.d/SPARK.CONF.DIR.d'
SPARK_JARS_IVY          = PROJECT_BASE_DIR + 'stack.d/spark.d/SPARK_JARS_IVY.d' # Cache downloaded JARS here.
SPARK_JARS_REPOSITORIES = ','.join([ 'http://repo.hortonworks.com/content/repositories/releases', ])

os.environ.pop('SPARK_MASTER_HOST', None)      # Since we're using pip/pySpark, these next three environment
os.environ.pop('SPARK_MASTER_POST', None)      # variables aren't needed; and we ensure pySpark doesn't
os.environ.pop('SPARK_HOME', None)             # get confused by them, should they be set.
os.environ.pop('PYTHONSTARTUP', None)          # Just in case pySpark 2.x attempts to read this.
os.environ['PYSPARK_PYTHON'] = sys.executable  # Make SPARK Workers/Executors use same Python as Master.
os.environ['JAVA_HOME'] = '/usr/lib/jvm/jre'   # Oracle JAVA for our pip/python3/pySpark 2.4 (not CDH's JRE).

os.environ['SPARK_CONF_DIR'] = SPARK_CONF_DIR  # Where spark-defaults.conf; spark-env.sh; log4j; etc. go.
os.environ['HADOOP_CONF_DIR'] = SPARK_CONF_DIR # Where symlinks to CDH's core|hdfs|hive|yarn-site.xml should
                                               # be created. (We use same directory as above). Both of these
                                               # locations can also be set in '(SPARK_CONF_DIR)/spark-env.sh'.

# ======================================================================
# Append CDH's Hadoop Native Library directory to LD_LIBRARY_PATH so
# Spark/PySpark can find them when interacting with HIVE/HADOOP/HDFS.
# Spark emits a warning when this path is absent, but it doesn't fail.
# ======================================================================
HADOOP_NATIVE_LIBS = '/opt/cloudera/parcels/CDH/lib/hadoop/lib/native/'
os.environ['LD_LIBRARY_PATH'] = os.environ.get('LD_LIBRARY_PATH', '') + ':' + HADOOP_NATIVE_LIBS
# ======================================================================

 
# ======================================================================
# Maven Coordinates for JARs (and their dependencies) needed to plug
# extra functionality into Spark 2.x.
#  -- spark-sql-kafka (Batch R/W from/to Kafka).
#  -- spark-streaming-kafka (Stream from/to Kafka).
#  -- Spark-Hbase-Connector from HortonWorks.
#      1) http://repo.hortonworks.com/content/repositories/releases/ # Released JAR and pom.xml deps.
#      2) https://github.com/hortonworks-spark/shc/                  # Source for latest and pom.xml deps.
# ======================================================================
SPARK_JARS_PACKAGES = ','.join([ 'org.apache.spark:spark-streaming-kafka-0-10_2.11:2.4.0',
                                 'org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0',
                                 'com.hortonworks:shc-core:1.1.1-2.1-s_2.11', ])
# ======================================================================
spark_conf = SparkConf()
spark_conf.setAll([
    ('spark.master', 'local[*]'),
    ('spark.app.name', 'myApp'),
    ('spark.submit.deployMode', 'client'),
    ('spark.ui.showConsoleProgress', 'true'),
    ('spark.eventLog.enabled', 'false'),
    ('spark.logConf', 'false'),
    ('spark.driver.bindAddress', '0.0.0.0'),
    ('spark.driver.host', '0.0.0.0'),

    ('spark.jars.repositories', SPARK_JARS_REPOSITORIES),
    ('spark.jars.ivy', SPARK_JARS_IVY),
    ('spark.jars.packages', SPARK_JARS_PACKAGES),

    ('spark.sql.warehouse.dir', SPARK_WAREHOUSE_DIR),
    ('spark.sql.catalogImplementation', 'hive'),
    ('spark.sql.hive.metastore.version', '1.2.1'),
   #('spark.sql.hive.metastore.version', '2.1.1'),
   #('spark.sql.hive.metastore.jars', '/path/to/2.1.1/jars'),

    # core|hive|hdfs|yarn-site.xml K/V pairs can be converted into equivalent SparkConf()
    # K/V pairs by prefixing their Keys w/ 'spark.hadoop.'. Also, symlinks to those CDH
    # files can be created in SPARK_CONF_DIR for Spark clients to use as (overrridable) defaults.
    ('spark.hadoop.fs.defaultFS', 'hdfs://vps00:8020'),                    # From: 'core-site.xml'
    ('spark.hadoop.hive.metastore.uris', 'thrift://vps00:9083'),           # From: 'hive-site.xml'
    ('spark.hadoop.hive.metastore.warehouse.dir', '/user/hive/warehouse'), # From: 'hive-site.xml'
])
 
spark_sesn          = SparkSession.builder.config(conf=spark_conf).enableHiveSupport().getOrCreate()
spark_ctxt          = spark_sesn.sparkContext
spark_reader        = spark_sesn.read
spark_streamReader  = spark_sesn.readStream
spark_ctxt.setLogLevel("WARN")
 
# ======================================================================
# Create small arbitrary DataFrame (myDF) and GroupedData (myGD)
# instances so we can inspect methods() and attributes inside them.
# Also, register table name for SparkSQL queries ('myDF_as_SQLtable').
# ======================================================================
myDF  = spark_sesn.createDataFrame([Row(0,1,2), Row(3,4,5), Row(6,7,8)]).toDF('col0', 'col1', 'col2')
myGDF = myDF.select('*').groupBy('col0')
myDF.createOrReplaceTempView('mydf_as_sqltable')
# ======================================================================
#spark_sens.stop()
# ======================================================================

100

import sys, os

from pyspark.conf import SparkConf

from pyspark.sql import SparkSession, Catalog

from pyspark.sql import DataFrame, DataFrameStatFunctions, DataFrameNaFunctions

from pyspark.sql import functions as F # Spark Functions. USAGE: F.col(), F.countDistinct()

from pyspark.sql import types as T # Spark Types. USAGE: T.Integer()

from pyspark.sql.types import Row

# ------------------------------------------

# Note: Row() in .../pyspark/sql/types.py

# isn't included in '__all__' list(), so

# we must import it by name here.

# ------------------------------------------

PROJECT_BASE_DIR = '/home/nmvega/agile.ds.d/'

SPARK_WAREHOUSE_DIR = PROJECT_BASE_DIR + 'stack.d/spark.d/spark-warehouse'

SPARK_CONF_DIR = PROJECT_BASE_DIR + 'stack.d/spark.d/SPARK.CONF.DIR.d'

SPARK_JARS_IVY = PROJECT_BASE_DIR + 'stack.d/spark.d/SPARK_JARS_IVY.d' # Cache downloaded JARS here.

SPARK_JARS_REPOSITORIES = ','.join([ 'http://repo.hortonworks.com/content/repositories/releases', ])

os.environ.pop('SPARK_MASTER_HOST', None) # Since we're using pip/pySpark, these next three environment

os.environ.pop('SPARK_MASTER_POST', None) # variables aren't needed; and we ensure pySpark doesn't

os.environ.pop('SPARK_HOME', None) # get confused by them, should they be set.

os.environ.pop('PYTHONSTARTUP', None) # Just in case pySpark 2.x attempts to read this.

os.environ['PYSPARK_PYTHON'] = sys.executable # Make SPARK Workers/Executors use same Python as Master.

os.environ['JAVA_HOME'] = '/usr/lib/jvm/jre' # Oracle JAVA for our pip/python3/pySpark 2.4 (not CDH's JRE).

os.environ['SPARK_CONF_DIR'] = SPARK_CONF_DIR # Where spark-defaults.conf; spark-env.sh; log4j; etc. go.

os.environ['HADOOP_CONF_DIR'] = SPARK_CONF_DIR # Where symlinks to CDH's core|hdfs|hive|yarn-site.xml should

# be created. (We use same directory as above). Both of these

# locations can also be set in '(SPARK_CONF_DIR)/spark-env.sh'.

# ======================================================================

# Append CDH's Hadoop Native Library directory to LD_LIBRARY_PATH so

# Spark/PySpark can find them when interacting with HIVE/HADOOP/HDFS.

# Spark emits a warning when this path is absent, but it doesn't fail.

# ======================================================================

HADOOP_NATIVE_LIBS = '/opt/cloudera/parcels/CDH/lib/hadoop/lib/native/'

os.environ['LD_LIBRARY_PATH'] = os.environ.get('LD_LIBRARY_PATH', '') + ':' + HADOOP_NATIVE_LIBS

# ======================================================================

# Maven Coordinates for JARs (and their dependencies) needed to plug

# extra functionality into Spark 2.x.

# -- spark-sql-kafka (Batch R/W from/to Kafka).

# -- spark-streaming-kafka (Stream from/to Kafka).

# -- Spark-Hbase-Connector from HortonWorks.

# 1) http://repo.hortonworks.com/content/repositories/releases/ # Released JAR and pom.xml deps.

# 2) https://github.com/hortonworks-spark/shc/ # Source for latest and pom.xml deps.

# ======================================================================

SPARK_JARS_PACKAGES = ','.join([ 'org.apache.spark:spark-streaming-kafka-0-10_2.11:2.4.0',

'org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0',

'com.hortonworks:shc-core:1.1.1-2.1-s_2.11', ])

# ======================================================================

spark_conf = SparkConf()

spark_conf.setAll([

('spark.master', 'local[*]'),

('spark.app.name', 'myApp'),

('spark.submit.deployMode', 'client'),

('spark.ui.showConsoleProgress', 'true'),

('spark.eventLog.enabled', 'false'),

('spark.logConf', 'false'),

('spark.driver.bindAddress', '0.0.0.0'),

('spark.driver.host', '0.0.0.0'),

('spark.jars.repositories', SPARK_JARS_REPOSITORIES),

('spark.jars.ivy', SPARK_JARS_IVY),

('spark.jars.packages', SPARK_JARS_PACKAGES),

('spark.sql.warehouse.dir', SPARK_WAREHOUSE_DIR),

('spark.sql.catalogImplementation', 'hive'),

('spark.sql.hive.metastore.version', '1.2.1'),

#('spark.sql.hive.metastore.version', '2.1.1'),

#('spark.sql.hive.metastore.jars', '/path/to/2.1.1/jars'),

# core|hive|hdfs|yarn-site.xml K/V pairs can be converted into equivalent SparkConf()

# K/V pairs by prefixing their Keys w/ 'spark.hadoop.'. Also, symlinks to those CDH

# files can be created in SPARK_CONF_DIR for Spark clients to use as (overrridable) defaults.

('spark.hadoop.fs.defaultFS', 'hdfs://vps00:8020'), # From: 'core-site.xml'

('spark.hadoop.hive.metastore.uris', 'thrift://vps00:9083'), # From: 'hive-site.xml'

('spark.hadoop.hive.metastore.warehouse.dir', '/user/hive/warehouse'), # From: 'hive-site.xml'

])

spark_sesn = SparkSession.builder.config(conf=spark_conf).enableHiveSupport().getOrCreate()

spark_ctxt = spark_sesn.sparkContext

spark_reader = spark_sesn.read

spark_streamReader = spark_sesn.readStream

spark_ctxt.setLogLevel("WARN")

# ======================================================================

# Create small arbitrary DataFrame (myDF) and GroupedData (myGD)

# instances so we can inspect methods() and attributes inside them.

# Also, register table name for SparkSQL queries ('myDF_as_SQLtable').

# ======================================================================

myDF = spark_sesn.createDataFrame([Row(0,1,2), Row(3,4,5), Row(6,7,8)]).toDF('col0', 'col1', 'col2')

myGDF = myDF.select('*').groupBy('col0')

myDF.createOrReplaceTempView('mydf_as_sqltable')

# ======================================================================

#spark_sens.stop()

# ======================================================================

Configure the DataFrameReader object. DataFrameReader objects offer a method to load various kinds of serialized formats (e.g. csv, json, parquet, etc) into a DataFrame object, as well as a method to set options related to that format. The following example is for a CSV file format:

spark_reader.format('csv')                             # CSV format.
spark_reader.option("inferSchema", "true")             # Infer schema from CSV.
spark_reader.option("header", "true")                  # CSV file has a header line.
spark_reader.option("ignoreLeadingWhiteSpace", "true") # Guard against spaces surrounding the delimiter.

spark_reader.format('csv') # CSV format.

spark_reader.option("inferSchema", "true") # Infer schema from CSV.

spark_reader.option("header", "true") # CSV file has a header line.

spark_reader.option("ignoreLeadingWhiteSpace", "true") # Guard against spaces surrounding the delimiter.

For this example, we’ll load real-estate sales data from Zillow into a pseudo-randomly named temporary filesystem file, which we’ll delete at the end. The file contents will be the data source for our resulting DAGs.

import requests

CSV_URL = 'https://fla.st/2PsNXsY'
# ===============================================================================
# Generate a temporary pseudo-random filesystem file and write the CSV-structured
# data to it. We'll use os.unlink() to delete it when we're done, but read the
# caution/gotcha note below which explains when we can delete it.
# ===============================================================================
with tempfile.NamedTemporaryFile(dir='/tmp', suffix='.csv', mode='r+', delete=False) as f:
    f.writelines(requests.get(CSV_URL).content.decode('utf-8')) 
    CSV_FILE = f.name
# ===============================================================================

df = spark_reader.load(CSV_FILE) # Read CSV file and create DataFrame.
df.show(5)                       # Preview with formatted output ...

# ===============================================================================
# We're tempted to delete the temporary file here, but can't! Why? Recall that
# Spark Lazily evaluates statements. Thus from here-on, every Action (i.e.
# outputs to console, persistence to external data-stores, or to Python-native
# data-structures) will trigger a DAG-flow which includes re-reading from the source
# data-store (i.e. our temporary file). We therefore cannot delete that file
# until the very end of this program. Further, we'll need to re-run this cell if
# we're re-running parts of this notebook and the file was already deleted.
# ===============================================================================
# os.unlink(CSV_FILE) # Run this statement ONLY at the end of this notebook/file!
# ===============================================================================

import requests

CSV_URL = 'https://fla.st/2PsNXsY'

# ===============================================================================

# Generate a temporary pseudo-random filesystem file and write the CSV-structured

# data to it. We'll use os.unlink() to delete it when we're done, but read the

# caution/gotcha note below which explains when we can delete it.

# ===============================================================================

with tempfile.NamedTemporaryFile(dir='/tmp', suffix='.csv', mode='r+', delete=False) as f:

f.writelines(requests.get(CSV_URL).content.decode('utf-8'))

CSV_FILE = f.name

# ===============================================================================

df = spark_reader.load(CSV_FILE) # Read CSV file and create DataFrame.

df.show(5) # Preview with formatted output ...

# ===============================================================================

# We're tempted to delete the temporary file here, but can't! Why? Recall that

# Spark Lazily evaluates statements. Thus from here-on, every Action (i.e.

# outputs to console, persistence to external data-stores, or to Python-native

# data-structures) will trigger a DAG-flow which includes re-reading from the source

# data-store (i.e. our temporary file). We therefore cannot delete that file

# until the very end of this program. Further, we'll need to re-run this cell if

# we're re-running parts of this notebook and the file was already deleted.

# ===============================================================================

# os.unlink(CSV_FILE) # Run this statement ONLY at the end of this notebook/file!

# ===============================================================================

+-----+-----------------------+-------+--------+-------+-------+-----------------+
|Index| "Living Space (sq ft)"| "Beds"| "Baths"|  "Zip"| "Year"| "List Price ($)"|
+-----+-----------------------+-------+--------+-------+-------+-----------------+
|  1.0|                 2222.0|    3.0|     3.5|32312.0| 1981.0|         250000.0|
|  2.0|                 1628.0|    3.0|     2.0|32308.0| 2009.0|         185000.0|
|  3.0|                 3824.0|    5.0|     4.0|32312.0| 1954.0|         399000.0|
|  4.0|                 1137.0|    3.0|     2.0|32309.0| 1993.0|         150000.0|
|  5.0|                 3560.0|    6.0|     4.0|32309.0| 1973.0|         315000.0|
+-----+-----------------------+-------+--------+-------+-------+-----------------+

+-----+-----------------------+-------+--------+-------+-------+-----------------+

+-----+-----------------------+-------+--------+-------+-------+-----------------+

| 1.0| 2222.0| 3.0| 3.5|32312.0| 1981.0| 250000.0|

| 2.0| 1628.0| 3.0| 2.0|32308.0| 2009.0| 185000.0|

| 3.0| 3824.0| 5.0| 4.0|32312.0| 1954.0| 399000.0|

| 4.0| 1137.0| 3.0| 2.0|32309.0| 1993.0| 150000.0|

| 5.0| 3560.0| 6.0| 4.0|32309.0| 1973.0| 315000.0|

+-----+-----------------------+-------+--------+-------+-------+-----------------+

Perform Transformations and Actions on this DataFrame, using either the DataFrame API or SQL statements. The following statements return DataFrames with identical contents and execution plans:

df.createOrReplaceTempView("df_asTable") # Register a SparkSQL table as name: 'df_asTable'
df_df  = df.filter(df.Zip == 32312) # .filter() and .where() are aliases.
df_sql = spark_sesn.sql(""" SELECT * FROM df_asTable WHERE Zip=32312 """)

df.createOrReplaceTempView("df_asTable") # Register a SparkSQL table as name: 'df_asTable'

df_df = df.filter(df.Zip == 32312) # .filter() and .where() are aliases.

df_sql = spark_sesn.sql(""" SELECT * FROM df_asTable WHERE Zip=32312 """)

print(id(df_df) == id(df_sql)) # Different DataFrame/Python objects
df_df.show()  # Display rows obtained via the DataFrame API.
df_sql.show() # Display rows obtained via SparkSQL statements.

print(id(df_df) == id(df_sql)) # Different DataFrame/Python objects

df_df.show() # Display rows obtained via the DataFrame API.

df_sql.show() # Display rows obtained via SparkSQL statements.

False

+-----+--------------------+----+-----+-----+--------------------+
|Index|Living Space (sq ft)|Beds|Baths|  Zip|Year, List Price ($)|
+-----+--------------------+----+-----+-----+--------------------+
|    1|                2222|   3|  3.5|32312|                1981|
|    3|                3824|   5|  4.0|32312|                1954|
|    6|                2893|   4|  3.0|32312|                1994|
|    8|                2483|   4|  3.0|32312|                2016|
|    9|                2400|   4|  4.0|32312|                2002|
|   12|                3200|   5|  4.0|32312|                1964|
+-----+--------------------+----+-----+-----+--------------------+

+-----+--------------------+----+-----+-----+--------------------+
|Index|Living Space (sq ft)|Beds|Baths|  Zip|Year, List Price ($)|
+-----+--------------------+----+-----+-----+--------------------+
|    1|                2222|   3|  3.5|32312|                1981|
|    3|                3824|   5|  4.0|32312|                1954|
|    6|                2893|   4|  3.0|32312|                1994|
|    8|                2483|   4|  3.0|32312|                2016|
|    9|                2400|   4|  4.0|32312|                2002|
|   12|                3200|   5|  4.0|32312|                1964|
+-----+--------------------+----+-----+-----+--------------------+

False

+-----+--------------------+----+-----+-----+--------------------+

+-----+--------------------+----+-----+-----+--------------------+

| 1| 2222| 3| 3.5|32312| 1981|

| 3| 3824| 5| 4.0|32312| 1954|

| 6| 2893| 4| 3.0|32312| 1994|

| 8| 2483| 4| 3.0|32312| 2016|

| 9| 2400| 4| 4.0|32312| 2002|

| 12| 3200| 5| 4.0|32312| 1964|

+-----+--------------------+----+-----+-----+--------------------+

+-----+--------------------+----+-----+-----+--------------------+

| 1| 2222| 3| 3.5|32312| 1981|

| 3| 3824| 5| 4.0|32312| 1954|

| 6| 2893| 4| 3.0|32312| 1994|

| 8| 2483| 4| 3.0|32312| 2016|

| 9| 2400| 4| 4.0|32312| 2002|

| 12| 3200| 5| 4.0|32312| 1964|

+-----+--------------------+----+-----+-----+--------------------+

Finally, we can delete the temporary file and shutdown the SparkSession …

if os.path.exists(CSV_FILE): os.unlink(CSV_FILE)
spark_sesn.stop()

1 2	if os.path.exists(CSV_FILE): os.unlink(CSV_FILE) spark_sesn.stop()

The end! =:)

ATTACHMENTS:
‣ Jupyter notebook of above session in HTML format