Author: Noelle Milton Vega
import multiprocessing
import pyspark
from pyspark.conf import SparkConf
from pyspark.sql import SparkSession
from pyspark.sql import functions as sFn # Usage: sFn.col(), sFn.window()
cpu_count = multiprocessing.cpu_count()
spark_uri = os.environ.get('SPARK_MASTER', 'local[%d]' % cpu_count)
spark_conf = SparkConf()
spark_conf.setAll( [('spark.master', spark_uri),
('spark.app.name', 'demoApp'),
('spark.submit.deployMode', 'client'),
('spark.ui.showConsoleProgress', 'true'),
('spark.eventLog.enabled', 'false'),
('spark.logConf', 'false')] )
spark_sesn = SparkSession.builder.config(conf=spark_conf).getOrCreate()
spark_ctxt = spark_sesn.sparkContext
spark_reader = spark_sesn.read # pyspark.sql.readwriter.DataFrameReader
spark_streamReader = spark_sesn.readStream # pyspark.sql.streaming.DataStreamReader
spark_ctxt.setLogLevel("INFO")
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.
import requests, tempfile
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!
# ===============================================================================
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.
if os.path.exists(CSV_FILE): os.unlink(CSV_FILE) # We can finally delete the temporary file.
spark_sesn.stop() # Shutdown Spark.
The end! =:)