<jupyter_start><jupyter_text>Exploring Heating Problems in ManhattanIn this notebook we will investigate New York City's 311 Complaints data, which is available as part of NYC Open Data. In particular we are interested in Heating Complaints within the Manhattan Borough. In this notebook, we will be exploring the data, using Spark's SQL module and the visualization tool Brunel.During the months of October to May, residents of NYC can call 311 to report that a building doesn't have enough heat or hot water. In the remaining months, June to September, complaints can be made that heating has been left on. There may be a number of different factors that contribute to heating complaints, we will select a few of the features available in our data to see if they suggest any correlation.*** Read the Data<jupyter_code>from pyspark.sql import SparkSession
import brunel
spark = SparkSession.builder.appName('NewYorkCase').getOrCreate()
nyc311DF=spark.read.csv('../datasets/311_Service_Requests_from_2015_subset.csv',header=True)
nyc311DF.count()
nyc311DF.printSchema()<jupyter_output>root
 |-- UniqueKey: string (nullable = true)
 |-- CreatedDate: string (nullable = true)
 |-- ClosedDate: string (nullable = true)
 |-- Agency: string (nullable = true)
 |-- AgencyName: string (nullable = true)
 |-- ComplaintType: string (nullable = true)
 |-- Descriptor: string (nullable = true)
 |-- LocationType: string (nullable = true)
 |-- IncidentZip: string (nullable = true)
 |-- IncidentAddress: string (nullable = true)
 |-- StreetName: string (nullable = true)
 |-- CrossStreet1: string (nullable = true)
 |-- CrossStreet2: string (nullable = true)
 |-- IntersectionStreet1: string (nullable = true)
 |-- IntersectionStreet2: string (nullable = true)
 |-- AddressType: string (nullable = true)
 |-- City: string (nullable = true)
 |-- Landmark: string (nullable = true)
 |-- FacilityType: string (nullable = true)
 |-- Status: string (nullable = true)
 |-- DueDate: string (nullable = true)
 |-- ResolutionDescription: string (nullable = true)
 |-- ResolutionActionUpdatedDate: string (n[...]<jupyter_text>Spark SQL ExplorationSparkSQL is a powerful tool allowing users a (often) familiar and (relatively) intuitive way to explore the data. In order to refer to the data within an SQL query, it needs to be stored as a view. The below query creates a temporary view named *nyc311ct*<jupyter_code>nyc311DF.createOrReplaceTempView("nyc311ct")<jupyter_output><empty_output><jupyter_text>We wish to limit our data to Manhattan, let's see the exact Borough value to use<jupyter_code>spark.sql("select distinct Borough from nyc311ct").show()<jupyter_output>+-------------+
|      Borough|
+-------------+
|  Unspecified|
|       QUEENS|
|     BROOKLYN|
|        BRONX|
|    MANHATTAN|
|STATEN ISLAND|
+-------------+<jupyter_text>Let's find the complaint type with the most complaints in Manhattan.Note that we are calling the cache function, this means that when the next action is called ("show", "count", etc.) it will store the dataframe *nyc311Agr_df* in memory for much quicker retrieval in the future. However, this must be small enough to fit.<jupyter_code>nyc311Agr_df = spark.sql("select `Complaint Type` as Complaint_Type, count(`Unique Key`) as Complaint_Count "+
                            "from nyc311ct where Borough = 'MANHATTAN' "+
                            "group by `Complaint Type` order by Complaint_Count desc").cache()
nyc311Agr_df.show()<jupyter_output>+--------------------+---------------+
|      Complaint_Type|Complaint_Count|
+--------------------+---------------+
|Noise - Street/Si...|            286|
|  Noise - Commercial|            164|
|     Illegal Parking|            162|
|            Graffiti|            104|
|             Smoking|            103|
|      Taxi Complaint|            102|
|Non-Residential Heat|             85|
|  Consumer Complaint|             73|
|   Broken Muni Meter|             68|
|     Noise - Vehicle|             62|
| Homeless Encampment|             59|
|    Street Condition|             58|
|  Food Establishment|             45|
|      HEAT/HOT WATER|             42|
|Fire Safety Direc...|             38|
|Maintenance or Fa...|             35|
|  Indoor Air Quality|             35|
|UNSANITARY CONDITION|             31|
|             Vending|             30|
|    Blocked Driveway|             29|
+--------------------+---------------+
only showing top 20 rows<jupyter_text>Visualize with BrunelConvert spark dataframe to Pandas Dataframe as brunel supports Pandas Dataframe show directly<jupyter_code>nyc311Agr_df_pd = nyc311Agr_df.toPandas()<jupyter_output><empty_output><jupyter_text>Let's get a visual representation of the data within nyc311Agr_df. We are creating a bubble chart, where the size of the bubble represents the number of complaints. The complaint type is assigned a color, and, if large enough, the bubble is labeled, else the type displayed when hovered over.<jupyter_code>%brunel data('nyc311Agr_df_pd') bubble size(Complaint_Count) color(Complaint_Type) label(Complaint_Type) legends(none) tooltip(Complaint_Type)<jupyter_output><empty_output><jupyter_text>Complaints by Zip codeHow does the number of complaints vary by Zip code? Let's remove any data points where a zip code hasn't been provided and filter to those that are of type 'HEAT/HOT WATER'.Note: If just exploring the data, where you do not intend to re-use the resulting dataframe, you can just use Spark SQL with the function "show" without assigning it to a variable.<jupyter_code>spark.sql("select `Incident Zip` as Zip, count(*) as ZipHeatingCnt " +
          "from nyc311ct " +
          "where `Complaint Type` = 'HEAT/HOT WATER' and `Incident Zip` <> '' group by `Incident Zip`").show()<jupyter_output>+-----+-------------+
|  Zip|ZipHeatingCnt|
+-----+-------------+
|11236|            2|
|11106|            2|
|10452|            1|
|11237|            1|
|11379|            1|
|11249|            3|
|10012|            1|
|11427|            1|
|11239|            1|
|11213|            6|
|10457|            1|
|11226|            6|
|11224|            1|
|10011|            1|
|10473|            1|
|11229|            1|
|11230|            1|
|10014|            1|
|10463|            1|
|10462|            1|
+-----+-------------+
only showing top 20 rows<jupyter_text>Similarly, if you wish to use the result of these queries for future queries but do not require the data as a dataframe, you can create a table directly from the query as follows"<jupyter_code>spark.sql("select `Incident Zip` as Zip, count(*) as ZipHeatingCnt " + 
          "from nyc311ct " +
          "where `Complaint Type` = 'HEAT/HOT WATER' and `Incident Zip` <> '' group by `Incident Zip`").createOrReplaceTempView("zipHeatingCnt")<jupyter_output><empty_output><jupyter_text>Additional SQL functions are available such as "max", "min", "cast", "split", "limit", etc. Some will be used below:<jupyter_code>spark.sql("select max(ZipHeatingCnt) as max, min(ZipHeatingCnt) as min, avg(ZipHeatingCnt) as avg from zipHeatingCnt").show()<jupyter_output>+---+---+------------------+
|max|min|               avg|
+---+---+------------------+
|  7|  1|2.0135135135135136|
+---+---+------------------+<jupyter_text>Complaints by Zip code and dateLet's see which date and zip codes had the most complaints. The 'Created Date' field includes a time, therefore we are using the "split" function to just use the date. We are also limiting the data to only heat/hot water complaints.<jupyter_code>spark.sql("select split(`Created Date`, ' ')[0] as Incident_Date, `Incident Zip` as Incident_Zip, "+
          "count(`Unique Key`) as HeatingComplaintCount "+
          "from nyc311ct where `Complaint Type` = 'HEAT/HOT WATER' and `Incident Zip` <> '' "+
          "group by split(`Created Date`, ' ')[0], `Incident Zip` order by HeatingComplaintCount desc limit 50").show()<jupyter_output>+-------------+------------+---------------------+
|Incident_Date|Incident_Zip|HeatingComplaintCount|
+-------------+------------+---------------------+
|   10/27/2015|       11212|                    2|
|   11/09/2015|       11212|                    2|
|   10/19/2015|       10456|                    2|
|   11/08/2015|       10456|                    2|
|   10/24/2015|       11209|                    2|
|   10/17/2015|       10467|                    1|
|   10/19/2015|       11211|                    1|
|   10/19/2015|       11237|                    1|
|   10/10/2015|       10451|                    1|
|   10/19/2015|       10011|                    1|
|   10/19/2015|       11373|                    1|
|   10/06/2015|       11207|                    1|
|   12/22/2015|       11213|                    1|
|   10/18/2015|       10453|                    1|
|   11/09/2015|       10021|                    1|
|   10/19/2015|       10024|                    1|
|   11/08/2015|       10454|   [...]<jupyter_text>It looks like Jan 8th and Feb 16th were cold days! Complaints by addressLet's explore complaints by address:<jupyter_code>addressNycAgr = spark.sql("select `Incident Address` as IncidentAddress, count(`Unique Key`) as IncidentCount "+
                              "from nyc311ct "+
                              "where `Complaint Type` = 'HEAT/HOT WATER' and `Incident Zip` <> '' group by `Incident Address`")
addressNycAgr.createOrReplaceTempView("nycaddHeat")<jupyter_output><empty_output><jupyter_text>Count how many addresses have complained about heating / hot water<jupyter_code>addressNycAgr.count()<jupyter_output><empty_output><jupyter_text>Let's investigate the distribution of the data; we tally up the number of addresses for each incident count.<jupyter_code>adrHeatDistr = spark.sql("select IncidentCount, count(*) as AddressDistr from nycaddHeat group by IncidentCount order by IncidentCount limit 100")
adrHeatDistr.createOrReplaceTempView("nycaddHeatDistr")
adrHeatDistr.show()<jupyter_output>+-------------+------------+
|IncidentCount|AddressDistr|
+-------------+------------+
|            1|         141|
|            2|           4|
+-------------+------------+<jupyter_text>Let's use Brunel again to get a visualization in the hope it gives up a quick understanding of the data.<jupyter_code>adrHeatDistr_pd = adrHeatDistr.toPandas()
%brunel data('adrHeatDistr_pd') x(IncidentCount) y(AddressDistr)<jupyter_output><empty_output><jupyter_text>The query took the first 100 incident counts but it looks like we have a long tail distribution; the bulk of the addresses only have 1 or a few complaints. The number of addresses that complained tends to decrease as the number of incidents counted increases.Let's have a quick look at the top number of complaints and how many addresses made that many:<jupyter_code>spark.sql("select * from nycaddHeatDistr order by IncidentCount desc limit 5").show()<jupyter_output>+-------------+------------+
|IncidentCount|AddressDistr|
+-------------+------------+
|            2|           4|
|            1|         141|
+-------------+------------+<jupyter_text>As expected, only one or two addresses can be found making each of number of complaints  Combining with PLUTO DataWe are now going to join the NYC 311 Complaints data with another data set, *PLUTO*. This data is also publically available with NYC Open Data. It is available in a CSV format and contains land use and geographic data, including more than seventy fields. If we can join these two datasets, we might be able to get a better understanding of attributes that contribute to complaints being made. We now read similar to before:<jupyter_code>plutoDF=spark.read.csv('../datasets/MN.csv',header=True)<jupyter_output><empty_output><jupyter_text>Let's confirm it worked by printing the schema<jupyter_code>plutoDF.printSchema()<jupyter_output>root
 |-- Borough: string (nullable = true)
 |-- Block: string (nullable = true)
 |-- Lot: string (nullable = true)
 |-- CD: string (nullable = true)
 |-- CT2010: string (nullable = true)
 |-- CB2010: string (nullable = true)
 |-- SchoolDist: string (nullable = true)
 |-- Council: string (nullable = true)
 |-- ZipCode: string (nullable = true)
 |-- FireComp: string (nullable = true)
 |-- PolicePrct: string (nullable = true)
 |-- Address: string (nullable = true)
 |-- ZoneDist1: string (nullable = true)
 |-- ZoneDist2: string (nullable = true)
 |-- ZoneDist3: string (nullable = true)
 |-- ZoneDist4: string (nullable = true)
 |-- Overlay1: string (nullable = true)
 |-- Overlay2: string (nullable = true)
 |-- SPDist1: string (nullable = true)
 |-- SPDist2: string (nullable = true)
 |-- LtdHeight: string (nullable = true)
 |-- AllZoning1: string (nullable = true)
 |-- AllZoning2: string (nullable = true)
 |-- SplitZone: string (nullable = true)
 |-- BldgClass: string (nullable = true)
 |--[...]<jupyter_text>We create a temporary view to be able to run SQL queries<jupyter_code>plutoDF.createOrReplaceTempView("plutot")<jupyter_output><empty_output><jupyter_text>Data Transformation & Feature EngineeringSometimes a feature needs to be engineered to make it more valuable, whether for visual purposes or modeling purposes. This can be done in many ways, including aggregation, bucketing, combining multiple features.Rather than investigate the data at an address level, we are going to aggregate to zipcode. We are also only selecting a subset of these fields to focus our investigation. The features we are extracting are Borough, ZipCode and the average for: Building Area, Building Age (by subtracting the build year from 2016), and X & Y co-ordinates.<jupyter_code>plutoSubdf = spark.sql("select Borough, ZipCode, count(*) as BldgCount, "+
                           "round(Avg(BldgArea)) as avgBldgArea, round(Avg(2015 - YearBuilt)) as avgBldgAge, "+
                           "round(Avg(XCoord)) as avgXcrd, round(Avg(YCoord)) as avgYcrd "+
                           "from plutot group by Borough, ZipCode")
plutoSubdf.createOrReplaceTempView("plutosubt")<jupyter_output><empty_output><jupyter_text>Merge the DataWe are now joining the NYC 311 complaints data with the PLUTO data where the two Zip codes match<jupyter_code>mergedDf = spark.sql("select * from zipHeatingCnt, plutosubt where Zip = ZipCode").cache()
mergedDf.createOrReplaceTempView("modelt")
mergedDf.count()<jupyter_output><empty_output><jupyter_text>We have only 46 Zip codes in Manhattan that made complaints in 2015, let's have a look at some of the attributes we have calculated<jupyter_code>spark.sql("select * from modelt").show(100)<jupyter_output>+-----+-------------+-------+-------+---------+-----------+----------+---------+--------+
|  Zip|ZipHeatingCnt|Borough|ZipCode|BldgCount|avgBldgArea|avgBldgAge|  avgXcrd| avgYcrd|
+-----+-------------+-------+-------+---------+-----------+----------+---------+--------+
|10012|            1|     MN|  10012|     1168|    23483.0|     143.0| 984755.0|203548.0|
|10011|            1|     MN|  10011|     2080|    26692.0|     145.0| 984425.0|209172.0|
|10014|            1|     MN|  10014|     1959|    17391.0|     148.0| 982940.0|206799.0|
|10463|            1|     MN|  10463|      220|    15722.0|     204.0|1008961.0|258548.0|
|10003|            3|     MN|  10003|     1881|    28940.0|     139.0| 987540.0|205715.0|
|10040|            1|     MN|  10040|      429|    40602.0|     188.0|1003705.0|251912.0|
|10022|            3|     MN|  10022|     1101|    82191.0|     189.0| 993309.0|215562.0|
|10065|            1|     MN|  10065|     1052|    42394.0|     127.0| 993913.0|217993.0|
|10035|   [...]<jupyter_text>Let's select the desired target (ZipHeatingCnt) and features (Borough, BldgCount, avgBldAge, avgBldgArea, avgXCrd, avgYCrd) to a dataframe ready to be saved for future use.<jupyter_code>mdlDataDF = spark.sql("select Zip, ZipHeatingCnt, Borough, BldgCount, avgBldgAge, avgBldgArea, avgXcrd, avgYCrd from modelt")<jupyter_output><empty_output><jupyter_text>Exploring Complaint DensityRather than look at complaint value as an absolute value, we are calculating and exploring Complaint Density by dividing the number of complaints by the number of buildings in that zipcode.We are also bucking the ages to see how the age of the building relates to the number of complaints.<jupyter_code>complaintDensityZipCode = spark.sql("select Borough, Zip, round((ZipHeatingCnt/BldgCount), 1) as ComplaintDensity, "+
                                        "round(avgBldgArea) as AvgBldgArea, round(avgBldgAge) as AvgBldgAge, "+
                                        "round(avgXcrd) as AvgXCoord, round(avgYcrd) as AvgYCoord from modelt").cache()<jupyter_output><empty_output><jupyter_text>Note: The "false" states that the results should not be truncated<jupyter_code>complaintDensityZipCode.show()
complaintDensityZipCode.createOrReplaceTempView("cdzt")<jupyter_output><empty_output><jupyter_text>We are once again utilizing Spark SQL's rich functionality and creating case statements to assign the zip code into either either *\<100* if the building is less than 100 years old, *100\<age\<200* if between 100 and 200 and *200+* if greater than 200.Similarly, the complaint density has been classified as *Low* if less than 0.1, H*igh* if between 0.1 and 1 and *Very High* if greater than 1.<jupyter_code>cdztop = spark.sql("select (case when AvgBldgAge < 100 then '<100' when AvgBldgAge >= 100 and AvgBldgAge < 200 then '100<age<200' else '>200' end) "+
                           "as AvgBldgAgeCat, "+
                       "(case when ComplaintDensity < 0.1 then 'Low' when ComplaintDensity >= 0.1 and ComplaintDensity < 1 then 'High' else 'Very High' end) "+
                           "as ComplaintDensityLevel "+
                       "from cdzt")<jupyter_output><empty_output><jupyter_text>Let's use the Chord graphic to visually interpret any connections between age and complaint density<jupyter_code>cdztop_pd = cdztop.toPandas()
%brunel  data('cdztop_pd') chord x(AvgBldgAgeCat) y(ComplaintDensityLevel) color(#count) tooltip(#all)<jupyter_output><empty_output><jupyter_text>There is only one zipcode with average building age less than 100 years old so this segmentation could be improved, or more data would be required to get a further understanding. However, as expected, the houses that are greater than 200 years old have a tendency to have a complaint density level or either High or Very High.  Saving the DataWhen saving, Spark will partition the data. It is worth understading how many rows are in your data as you can explicitly specify how many partitions you would like created<jupyter_code>mdlDataDF.count()<jupyter_output><empty_output><jupyter_text>Since we only have a small amount of data we can keep it all together, we do this by calling "coalesce" and specifying the value *1*<jupyter_code>mergedMNcmp = mdlDataDF.coalesce(1)<jupyter_output><empty_output><jupyter_text>Similar to how we specified there was a header in the raw file, we pass this parameter to "options" after calling "write". We specify that we wish for the data to be saved in a csv format by passing the appropriate path and filename to the "csv" method.<jupyter_code>mergedMNcmp.write.csv('../datasets/mnModelData.csv',header=True,mode="overwrite")<jupyter_output><empty_output><jupyter_text>We repeat for the subset of pluto data<jupyter_code>plutoSubdf.count()
plutoSubcmp = plutoSubdf.coalesce(1)
plutoSubcmp.write.csv('../datasets/mnAgrData.csv',header="true",mode="overwrite")
!ls -l ../datasets<jupyter_output>total 19230
-rwx------. 1 999 root  6710815 Aug  1 18:46 311_Service_Requests_from_2015_subset.csv
drwxrwxrwx. 2 999 999      4096 Aug  1 19:22 mnAgrData.csv
-rwx------. 1 999 root 12972032 Aug  1 19:17 MN.csv
drwxrwxrwx. 2 999 999      4096 Aug  1 19:22 mnModelData.csv
{'fraction_non_alphanumeric': 0.1303951367781155, 'fraction_numerical': 0.052077001013171226, 'mean_word_length': 2.908334983171649, 'pattern_counts': {'":': 0, '<': 124, '<?xml version=': 0, '>': 117, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '30', 'dir': 'jupyter-structured-clean-dedup', 'id': '538632', 'license': 'Apache-2.0', 'n_tokens_mistral': 6782, 'n_tokens_neox': 5912, 'n_words': 1971, 'path': 'notebooks/NewYork-Case-Data-Exploration.jupyter.ipynb', 'repo_name': 'ibm-cloud-architecture/nyc-heating-problems', 'size': '19740'}