<jupyter_start><jupyter_text>[Hive Partitioning with Spark](https://datascience-enthusiast.com/Python/hive_partitioning_with_Spark.htmlHive-Partitioning-with-Spark)[NYC Opendata 311 Service Requests from 2010 to Present](https://nycopendata.socrata.com/api/views/erm2-nwe9/rows.csv?accessType=DOWNLOAD)<jupyter_code>!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q http://www-eu.apache.org/dist/spark/spark-2.3.3/spark-2.3.3-bin-hadoop2.7.tgz
!tar xf spark-2.3.3-bin-hadoop2.7.tgz
!pip install -q findspark
!pip install pyspark
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.3.3-bin-hadoop2.7"
import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.enableHiveSupport().config("spark.network.timeout", '200s').getOrCreate()
from pyspark.sql.types import *
!wget "https://nycopendata.socrata.com/api/views/erm2-nwe9/rows.csv?accessType=DOWNLOAD"
!ls
!cat rows.csv\?accessType\=DOWNLOAD | head
!head -1 rows.csv?accessType=DOWNLOAD
columns = """Unique Key,Created Date,Closed Date,Agency,Agency Name,Complaint Type,Descriptor,Location Type,Incident Zip,Incident Address,Street Name,Cross Street 1,Cross Street 2,Intersection Street 1,Intersection Street 2,Address Type,City,Landmark,Facility Type,Status,Due Date,Resolution Description,Resolution Action Updated Date,Community Board,BBL,Borough,X Coordinate,Y Coordinate,Open Data Channel Type,Park Facility Name,Park Borough,Vehicle Type,Taxi Company Borough,Taxi Pick Up Location,Bridge Highway Name,Bridge Highway Direction,Road Ramp,Bridge Highway Segment,Latitude,Longitude,Location"""
columns = columns.replace(" ",'_').replace('\n','').lower().split(",")
columns[:10]  # just showing the first few column names
nyc_schema = StructType([StructField(colName, StringType(), True) for colName in columns])
print(nyc_schema)
nyc311 = spark.read.csv("rows.csv?accessType=DOWNLOAD",
                         header = True, 
                         schema = nyc_schema)
nyc311.select('created_date','agency_name','city').show(10, truncate = False)
nyc311.count()
nyc311.show(5)
from datetime import datetime
from pyspark.sql.functions import col,udf, year

to_date_time =  udf(lambda x: datetime.strptime(x, '%m/%d/%Y %I:%M:%S %p'), TimestampType())
nyc311_2 = nyc311.withColumn('created_date', to_date_time(col('created_date'))).withColumn('year', year('created_date'))
nyc311_2.select(['created_date', 'year']).show(10)
nyc311_2.sample(False, 0.5, 42).createOrReplaceTempView('nyc311_table')
create_partitioned_table_sql = """CREATE TABLE if not exists nyc311_orc_partitioned (unique_key STRING, created_date STRING, closed_date STRING, agency STRING, agency_name STRING, complaint_type STRING, descriptor STRING, location_type STRING, incident_zip STRING, incident_address STRING, street_name STRING, cross_street_1 STRING, cross_street_2 STRING, intersection_street_1 STRING, intersection_street_2  STRING, address_type STRING, city  STRING, landmark STRING, facility_type STRING, status STRING, due_date STRING, resolution_description  STRING,resolution_action_updated_date STRING, community_board STRING, bbl STRING, borough STRING, x_coordinate STRING, y_coordinate STRING, open_data_channel_type STRING, park_facility_name STRING, park_borough STRING, vehicle_type STRING, taxi_company_borough STRING, taxi_pick_up_location STRING, bridge_highway_name STRING, bridge_highway_direction  STRING, road_ramp STRING, bridge_highway_segment STRING, latitude STRING, longitude STRING, location STRING) PARTITIONED BY (year INT) STORED AS ORC """
spark.sql(create_partitioned_table_sql)
set_dynamic_mode  = "SET hive.exec.dynamic.partition.mode = nonstrict"
spark.sql(set_dynamic_mode)
insert_sql = """INSERT INTO nyc311_orc_partitioned PARTITION (year) SELECT unique_key, created_date, closed_date, agency, agency_name, complaint_type, descriptor, location_type, incident_zip, incident_address, street_name, cross_street_1, cross_street_2, intersection_street_1, intersection_street_2, address_type, city, landmark, facility_type, status, due_date, resolution_description, resolution_action_updated_date, community_board, bbl, borough, x_coordinate, y_coordinate, open_data_channel_type, park_facility_name, park_borough, vehicle_type, taxi_company_borough, taxi_pick_up_location, bridge_highway_name, bridge_highway_direction, road_ramp, bridge_highway_segment, latitude, longitude, location, year FROM nyc311_table"""
spark.sql(insert_sql)
spark.sql("show partitions nyc311_orc_partitioned").show()
spark.sql('create table nyc311_orc_unpartitioned stored as orc as select * from nyc311_table')
import os
os.listdir("spark-warehouse/nyc311_orc_unpartitioned")[:10]
os.listdir("spark-warehouse/nyc311_orc_partitioned")
os.listdir("spark-warehouse/nyc311_orc_partitioned/year=2010")[:10]
import time
t0 = time.time()
aggregates = spark.sql("select agency_name, city, incident_zip, count(*) as frequency from nyc311_orc_unpartitioned where year = '2010' group by agency_name, city, incident_zip").toPandas()
t1 = time.time()

print('time taken with unpartitioned table ', round(t1-t0,2))
t0 = time.time()
aggregates = spark.sql("select agency_name, city,incident_zip, count(*) as frequency from nyc311_orc_partitioned where year = '2010' group by agency_name, city, incident_zip").toPandas()
t1 = time.time()

print('time taken with partitioned table ', round(t1-t0,2))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11066471877282688, 'fraction_numerical': 0.027209642074506938, 'mean_word_length': 7.638801261829653, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 6, 'https://': 3, 'lorem ipsum': 0, 'www.': 1, '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': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '435158', 'license': 'MIT', 'n_tokens_mistral': 1927, 'n_tokens_neox': 1771, 'n_words': 390, 'path': 'Hive/Hive_Partitioning_with_Spark.ipynb', 'repo_name': 'Nik8x/PySpark', 'size': '5476'}