<jupyter_start><jupyter_text>Unit E SQL Over Anything with Drill and Spark- Examples From Video Lecture<jupyter_code>import pyspark
from pyspark.sql import SparkSession

bucket = "e-drill"

spark = SparkSession.builder \
    .master("local") \
    .appName('jupyter-pyspark') \
        .config("spark.jars.packages","org.apache.hadoop:hadoop-aws:3.1.2,org.apache.spark:spark-avro_2.12:3.1.2")\
        .config("spark.hadoop.fs.s3a.endpoint", "http://minio:9000") \
        .config("spark.hadoop.fs.s3a.access.key", "minio") \
        .config("spark.hadoop.fs.s3a.secret.key", "SU2orange!") \
        .config("spark.hadoop.fs.s3a.fast.upload", True) \
        .config("spark.hadoop.fs.s3a.path.style.access", True) \
        .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
    .getOrCreate()
sc = spark.sparkContext
sc.setLogLevel("ERROR") # Keeps the noise down!!!<jupyter_output>WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/usr/local/spark-3.1.2-bin-hadoop3.2/jars/spark-unsafe_2.12-3.1.2.jar) to constructor java.nio.DirectByteBuffer(long,int)
WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
Ivy Default Cache set to: /home/jovyan/.ivy2/cache
The jars for the packages stored in: /home/jovyan/.ivy2/jars
org.apache.hadoop#hadoop-aws added as a dependency
org.apache.spark#spark-avro_2.12 added as a dependency
:: resolving dependencies :: org.apache.spark#spark-submit-parent-1c59d893-fe2d-43cc-b4d7-1fefd4202193;1.0
	confs: [default]<jupyter_text>Setup- Put data in the right places!!!- Run these cells to ensure you have the data for the examples<jupyter_code>! pip install -q minio
from minio import Minio

# Make the minio bucket
client = Minio("minio:9000","minio","SU2orange!", secure=False)
not client.bucket_exists(bucket) and client.make_bucket(bucket)

# open the example(s)
gp = spark.read.json("file:///home/jovyan/datasets/json-samples/google-places.json")
c = spark.read.csv("file:///home/jovyan/datasets/customers/customers.csv", inferSchema=True, header=True)
s = spark.read.csv("file:///home/jovyan/datasets/customers/surveys.csv",inferSchema=True, header=True)
g = spark.read.csv("file:///home/jovyan/datasets/grades/*.tsv",inferSchema=False, header=False, sep="\t")

# Put the examples in minio
gp.write.mode("Overwrite").json(f"s3a://{bucket}/google-places.json")
c.write.mode("Overwrite").csv(f"s3a://{bucket}/customers.csv",header=True)
s.write.mode("Overwrite").csv(f"s3a://{bucket}/surveys.csv",header=True)
g.write.mode("Overwrite").csv(f"s3a://{bucket}/grades.csv",header=False, sep=",")
g.write.mode("Overwrite").parquet(f"s3a://{bucket}/grades.parquet")

#put the examples in HDFS
gp.write.mode("Overwrite").json(f"hdfs://namenode/user/root/{bucket}/google-places.json")
c.write.mode("Overwrite").csv(f"hdfs://namenode/user/root/{bucket}/customers.csv",header=True)
s.write.mode("Overwrite").csv(f"hdfs://namenode/user/root/{bucket}/surveys.csv",header=True)
g.write.mode("Overwrite").csv(f"hdfs://namenode/user/root/{bucket}/grades.csv",header=False, sep=",")
g.write.mode("Overwrite").parquet(f"hdfs://namenode/user/root/{bucket}/grades.parquet")<jupyter_output><empty_output><jupyter_text>Apache DrillThese Examples Use Apache Drill. Content is here for reference only. Drilling Object Storage / Minio```MINIO UIMake Bucket e-demoMake folder examscoresUpload file exam-scores.xlsxDrill Storage Config:Setup S3 storage plugin, call it s3edemoMake sure xlsx file extensions point to the correct driver.DRILL QUERIES:Select * from ,* from s3edemo.`excel/exam-scores.xlsx`select ntile(4) over (order by student_score desc) as ranking,* from s3demo.`excel/exam-scores.xlsx`select avg(Student_Score) over (partition by exam_version ) as avg_by_exam_version,avg(Student_Score) over( partition by class_section) as avg_by_class_section,* from s3demo.`excel/exam-scores.xlsx```` Drilling HDFS```Drill Storage Config: - use file config - set connection to hdfs://namenode:8020 - set root location to hdfs path /user/root/ - make sure you have formats correct for your file types  DRILL QUERIES: select * from hdfs.root.`e-drill/customers.csv`;select * from hdfs.root.`e-drill/grades.parquet`;``` Drilling Web Services over HTTP```Drill Storage config: Singlefile: STOCKS-----------------------------      "stocks": {        "url": "https://raw.githubusercontent.com/mafudge/datasets/master/stocks/stocks.json",        "requireTail": false,}DRILL QUERIES: select * from http.stocks;Drill Storage Config: a requireTail example-------------------------------------------Query all files here: https://github.com/mafudge/advanced-databases/tree/main/datasets/json-samples "jsonsamples" : {        "url" : "https://raw.githubusercontent.com/mafudge/advanced-databases/main/datasets/json-samples/",        "requireTail" : true    }DRILL QUERIES: select * from http.jsonsamples.`stocks.json`;select * from http.jsonsamples.`US-Senators.json`;select * from http.jsonsamples.`europe.json`;Drill Storage Config DATAPATH and PARAMETERS --------------------------------------------- "sunrise": {      "url": "https://api.sunrise-sunset.org/json", "params": [        "lat",        "lng",        "date"      ],      "dataPath": "results",      "requireTail": false },  DRILL QUERIES:select * from http.sunrise; Drill Storage Config: WEATHER API ---------------------------------- "weather_daily": {      "url": "https://api.openweathermap.org/data/2.5/onecall",      "method": "GET",      "headers": null,      "authType": "none",      "userName": null,      "password": <PASSWORD>,      "postBody": null,      "params": [        "lat",        "lon",        "appid",        "units"      ],      "dataPath": "daily",      "requireTail": false,      "inputType": "json"  }DRILL QUERIES: select * from http.weather_daily`;``` Creating Table Schemas```SELECT * FROM s3. `grades.csv`Whoops header is first row!Configure storage provider to not include headerSELECT * FROM s3. `grades.csv`Whoops one column names “COLUMNS” with an array of values-- parse the columnsSELECT COLUMNS[0], COLUMNS[1] FROM s3. `grades/*.tsv`--with names and typesSELECT cast(COLUMNS[0] as int) as year, COLUMNS[1] FROM s3..`grades/*.tsv`-- as a CTEwith grades as (    select cast(COLUMNS[0] as int) as year, COLUMNS[1] as semester,COLUMNS[2] as course,cast(COLUMNS[3] as int) as credits,COLUMNS[4] as letter_gradefrom s3.`grades/*.tsv`    )select * from grades``` Creating Tables and Views```See what's out there:SHOW SCHEMA or SHOW DATABASESWhich are writable?DESCRIBE SCHEMA s3create view s3.`vgrades` (year, semester,course, credits, gradee)  as  select * from  hdfs.root.`e-drill/grades.parquet`SHOW TABLES Select * from s3.vgradesdrop view  s3.vgrades Go check it out on minio... a little file with the query plan!!!Try to create a table from http.stockscreate table dfs.tmp.stocks asselect symbol, price from http.stocksSelect * from dfs.tmp.stocks``` Drilling Nested JSON```select g.name,     g.geometry.location.lat as `lat`,     g.geometry.location.lng as `lng`,     FLATTEN(g.types) as `type`,     * from s3.`google-places.json` as gselect s.`congress_numbers`[0], * from http.jsonsamples.`US-Senators.json` s```  Spark SQL- Spark Supports ANSI Standard SQL- Once you register a DataFrame as a Temp View, you can query it like an SQL table.- The output is a DataFrame, that you can maniuplate further with spark!<jupyter_code>c.createOrReplaceTempView("customers")
s.createOrReplaceTempView("surveys")
c.printSchema()
s.printSchema()
query = '''
select c.Email, c.Gender, c.State, c.`Months Customer`, s.`Own Home`, s.`Household Income`
from customers c left join surveys s on 
        c.Email = s.Email
    where c.State = 'NY'
    and c.`Months Customer` > 5
    and s.`Own Home` is not null
'''
df = spark.sql(query)
df.show()
df.printSchema()
c.createOrReplaceTempView("customers")
s.createOrReplaceTempView("surveys")
c.printSchema()
s.printSchema()

query = '''
select c.Email, c.Gender, c.State, c.`Months Customer`, s.`Own Home`, 
    case s.`Household Income` 
        when 'Prefer not to Answer' then null 
        else cast(s.`Household Income` as DOUBLE) 
    end as Income
from customers c left join surveys s on 
        c.Email = s.Email
    where c.State = 'NY'
    and c.`Months Customer` > 5
    and s.`Own Home` is not null
'''
df = spark.sql(query)
df.show()
df.printSchema()
df.explain()
df.createOrReplaceTempView("v_ny_customer_analysis")<jupyter_output>root
 |-- Email: string (nullable = true)
 |-- Gender: string (nullable = true)
 |-- State: string (nullable = true)
 |-- Months Customer: integer (nullable = true)
 |-- Own Home: string (nullable = true)
 |-- Income: double (nullable = true)

== Physical Plan ==
*(2) Project [Email#368, Gender#369, State#372, Months Customer#375, Own Home#406, CASE WHEN (Household Income#405 = Prefer not to Answer) THEN null ELSE cast(Household Income#405 as double) END AS Income#656]
+- *(2) BroadcastHashJoin [Email#368], [Email#402], Inner, BuildRight, false
   :- *(2) Filter ((((isnotnull(State#372) AND isnotnull(Months Customer#375)) AND (State#372 = NY)) AND (Months Customer#375 > 5)) AND isnotnull(Email#368))
   :  +- FileScan csv [Email#368,Gender#369,State#372,Months Customer#375] Batched: false, DataFilters: [isnotnull(State#372), isnotnull(Months Customer#375), (State#372 = NY), (Months Customer#375 > 5..., Format: CSV, Location: InMemoryFileIndex[file:/home/jovyan/datasets/customers/custome[...]<jupyter_text>Spark SQL works with Nested Data<jupyter_code>gp.printSchema()
gp.createOrReplaceTempView("googleplaces")
query = '''
select name, geometry.location.lat, geometry.location.lng, explode(types) as type from googleplaces
'''
spark.sql(query).show()<jupyter_output>+--------------------+-----------------+------------------+-----------------+
|                name|              lat|               lng|             type|
+--------------------+-----------------+------------------+-----------------+
|            Syracuse|       43.0481221|-76.14742439999999|         locality|
|            Syracuse|       43.0481221|-76.14742439999999|        political|
|Crowne Plaza Syra...|       43.0476078|       -76.1417642|          lodging|
|Crowne Plaza Syra...|       43.0476078|       -76.1417642|point_of_interest|
|Crowne Plaza Syra...|       43.0476078|       -76.1417642|    establishment|
|  The Parkview Hotel|       43.0476157|        -76.140986|          lodging|
|  The Parkview Hotel|       43.0476157|        -76.140986|point_of_interest|
|  The Parkview Hotel|       43.0476157|        -76.140986|    establishment|
|<NAME>...|       43.0472894|-76.15385049999999|          lodging|
|<NAME>...|       43.0472894|-76.15385049999999|point[...]<jupyter_text>Registering a UDF for use in Spark SQL<jupyter_code>from pyspark.sql.functions import udf
from pyspark.sql.types import * 

@udf(returnType=StringType()) 
def upperCase(str):
    return str.upper()

spark.udf.register("upperCase", upperCase)

spark.sql("select Email, upperCase(Email), Gender from v_ny_customer_analysis").show()<jupyter_output>+--------------------+--------------------+------+
|               Email|           uc(Email)|Gender|
+--------------------+--------------------+------+
|<EMAIL>...|ETASOMTHIN@SUP<EMAIL>...|     M|
|   <EMAIL>|   <EMAIL>|     F|
| <EMAIL>| <EMAIL>|     M|
| <EMAIL>| <EMAIL>|     M|
| <EMAIL>| <EMAIL>|     M|
+--------------------+--------------------+------+<jupyter_text>Listing views in Your Session<jupyter_code>spark.sql("show views").show()<jupyter_output>+---------+--------------------+-----------+
|namespace|            viewName|isTemporary|
+---------+--------------------+-----------+
|         |           customers|       true|
|         |        googleplaces|       true|
|         |             surveys|       true|
|         |v_ny_customer_ana...|       true|
+---------+--------------------+-----------+<jupyter_text>Big Data To Small DataUsing the `.toPandas()` function will materilaize the spark DataFrame as a Python Pandas dataframe. This allows us to use small-data features available in Python as our code is no longer executing on the cluster. Graphing Output- A Spark DataFrame can be converted to a Python Pandas dataframe with the `toPandas()` function.- Once its in Pandas you can plot with standard plotting libraries.- https://pandas.pydata.org/pandas-docs/stable/user_guide/visualization.html<jupyter_code>c.toPandas().plot.bar(x="Email", y="Total Purchased")<jupyter_output><empty_output><jupyter_text>IPywidgets- IPy Widgets allows us to create interactive notebooks- https://ipywidgets.readthedocs.io/en/latest/- This is a great example of going from "Big Data" to "Small Data"- As a rule, we don't want to produce small data (Pandas DataFrame) until we have mapped or reduced results.<jupyter_code>from IPython.display import display, HTML
from ipywidgets import interact, interact_manual
c = spark.read.csv("file:///home/jovyan/datasets/customers/customers.csv", inferSchema=True, header=True)

display(HTML("<h1>Customers By State</h1>"))
states = c.select("State").distinct().toPandas()["State"].values
states.sort()
@interact(state=states)
def main(state :str):
    rows = c.where( c.State == state).toPandas()
    display(rows)
from IPython.display import display, HTML
from ipywidgets import interact, interact_manual, widgets

display(HTML("<h1>Customers By Total Purchased</h1>"))
rangewidget = widgets.IntRangeSlider(
    value=[10, 1000],
    min=0,
    max=5000,
    step=10,
    description='Total Purchased:',
    disabled=False,
    continuous_update=True,
    orientation='horizontal',
    readout=True,
    readout_format='d',
)

c.createOrReplaceTempView("customers")

@interact_manual(purchased=rangewidget)
def main(purchased: int):
    query = f'select * from customers where `Total Purchased` between {purchased[0]} and {purchased[1]}'
    display(spark.sql(query).toPandas())
    
s.printSchema()
s.orderBy("Favorite Department", ascending = False).show()

s.orderBy(s["Favorite Department"],ascending = False).show()
s.filter( "`Favorite Department` LIKE 'Prefer%' " ).show()

from pyspark.sql.functions import *
s.filter(col("Favorite Department").startswith("Prefer") ).show()
s.createOrReplaceTempView("surveys")
spark.sql("show tables").show()
z =spark.sql("select * from surveys where `Favorite Department` like 'Prefer%'")
y = s.filter(col("Favorite Department").startswith("Prefer") )
z =spark.sql("select * from surveys where `Favorite Department` like 'Prefer%'")
y.explain()
z.explain()
query = '''

SELECT *
    FROM  surveys
    WHERE `Favorite Department` like 'Prefer%'

'''

spark.sql(query)
#pyspark
spark.read.csv("s3a://....")
spark.read.format("csv").load("s3a:///...")
s.write.format("foo.bvar.avro").mode("errorifexists")\
    .option("sep","-").option("header",True).save("file:///foo.csv")
gp.select("geometry.location.lat",gp.geometry.location.lng.alias("lng")  ).show()
gp.select(col("geometry.location.lat").alias("g.l.lat"),gp.geometry.location.lng.alias("g.l.lng")  ).show()
gp.select("geometry.location.lat",gp.geometry.location.lng ).toDF("g.l.lat","g.l.lng").show()
from pyspark import SparkFiles
spark.sparkContext.addFile("https://raw.githubusercontent.com/mafudge/datasets/master/crime/nys-crime-1960-2012.csv")
df = spark.read.option("inferSchema",True).option("header", True)\
    .csv(SparkFiles.get("nys-crime-1960-2012.csv"))
df.printSchema()
SparkFiles.get("nys-crime-1960-2012.csv")
from IPython.display import display, HTML
from ipywidgets import interact, interact_manual


@interact_manual(name="", age=(1,99), retired=False, hobbies=["Biking","Skiing"])
def onclick(name,age,retired,hobbies):
    display(HTML(f"<p>{name}, {age} {retired} {hobbies}"))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.17978159126365054, 'fraction_numerical': 0.02683307332293292, 'mean_word_length': 3.6304536261196185, 'pattern_counts': {'":': 20, '<': 46, '<?xml version=': 0, '>': 50, 'https://': 8, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '559035', 'license': 'Apache-2.0', 'n_tokens_mistral': 5653, 'n_tokens_neox': 5254, 'n_words': 1192, 'path': 'work/content/E-Drill-SparkSQL.ipynb', 'repo_name': 'mafudge/advanced-databases', 'size': '16139'}