<jupyter_start><jupyter_text>Working with Data Frame API+ 1.Schema+ 2.Working with Columns+ 3.Working with Rows+ 3.Challenges---- dataset: https://data.cityofchicago.org/Public-Safety/Crimes-2001-to-Present/ijzp-q8t2- click Export - choose 'CSV'  Set up<jupyter_code>from pyspark import SparkContext
sc = SparkContext.getOrCreate()
import pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
spark<jupyter_output><empty_output><jupyter_text>Preprocess Chicago's reported Crimes Data<jupyter_code>from pyspark.sql.functions import to_timestamp, col, lit
rc = spark.read.csv('../data/chicago_crimes.csv', header=True).withColumn('Date', to_timestamp(col('Date'),'MM/dd/yyyy hh:mm:ss a')).filter(col('Date') <= lit('2018-11-11'))
rc.show(5)
rc.show(1)<jupyter_output>+--------+-----------+-------------------+---------------+----+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|      ID|Case Number|               Date|          Block|IUCR|Primary Type|         Description|Location Description|Arrest|Domestic|Beat|District|Ward|Community Area|FBI Code|X Coordinate|Y Coordinate|Year|          Updated On|    Latitude|    Longitude|            Location|
+--------+-----------+-------------------+---------------+----+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|10224738|   HY411648|2015-09-05 13:30:00|043XX S WOOD ST|0486|     BATTERY|DOMESTIC BATTERY ...|           RESIDENCE| false|    true|0924|     009|  12|            6[...]<jupyter_text>--------  1) Schemas  print current schema<jupyter_code># check current schema
rc.printSchema()<jupyter_output>root
 |-- ID: string (nullable = true)
 |-- Case Number: string (nullable = true)
 |-- Date: timestamp (nullable = true)
 |-- Block: string (nullable = true)
 |-- IUCR: string (nullable = true)
 |-- Primary Type: string (nullable = true)
 |-- Description: string (nullable = true)
 |-- Location Description: string (nullable = true)
 |-- Arrest: string (nullable = true)
 |-- Domestic: string (nullable = true)
 |-- Beat: string (nullable = true)
 |-- District: string (nullable = true)
 |-- Ward: string (nullable = true)
 |-- Community Area: string (nullable = true)
 |-- FBI Code: string (nullable = true)
 |-- X Coordinate: string (nullable = true)
 |-- Y Coordinate: string (nullable = true)
 |-- Year: string (nullable = true)
 |-- Updated On: string (nullable = true)
 |-- Latitude: string (nullable = true)
 |-- Longitude: string (nullable = true)
 |-- Location: string (nullable = true)<jupyter_text>Create customized schema- column name, variable type, nullable<jupyter_code>from pyspark.sql.types import StructType, StructField, StringType, TimestampType, BooleanType, DoubleType, IntegerType
# get the columns
rc.columns
len(rc.columns)<jupyter_output><empty_output><jupyter_text>Option 1) manully creating StructType<jupyter_code># column name, variable type, nullable 
schema = StructType([
    StructField('ID', StringType(), True),
    StructField('Case Number', StringType(), True),
    StructField('Date', TimestampType(), True),
    StructField('Block',StringType(), True),
    StructField('IUCR',StringType(), True),
    StructField('Primary Type', StringType(), True),
    StructField('Description', StringType(), True),
    StructField('Location Description', StringType(), True),
    StructField('Arrest', StringType(), True),
    StructField('Domestic', BooleanType(), True),
    StructField('Beat', StringType(), True),
    StructField('District', StringType(), True),
    StructField('Ward', StringType(), True),
    StructField('Community Area', StringType(), True),
    StructField('FBI Code', StringType(), True),
    StructField('X Coordinate', StringType(), True),
    StructField('Y Coordinate', StringType(), True),
    StructField('Year', IntegerType(), True),
    StructField('Updated On', StringType(), True),
    StructField('Latitude', DoubleType(), True),
    StructField('Longitude', DoubleType(), True),
    StructField('Location', StringType(), True),
])

schema<jupyter_output><empty_output><jupyter_text>Option 2) creating Struct Field using Python tuples<jupyter_code>labels = [
    ('ID',StringType()),
    ('Case Number',StringType()),
     ('Date',TimestampType()),
     ('Block',StringType()),
     ('IUCR',StringType()),
     ('Primary Type',StringType()),
     ('Description',StringType()),
     ('Location Description',StringType()),
     ('Arrest',StringType()),
     ('Domestic',BooleanType()),
     ('Beat',StringType()),
     ('District',StringType()),
     ('Ward',StringType()),
     ('Community Area',StringType()),
     ('FBI Code',StringType()),
     ('X Coordinate',StringType()),
     ('Y Coordinate',StringType()),
     ('Year',IntegerType()),
     ('Updated On',StringType()),
     ('Latitude',DoubleType()),
     ('Longitude',DoubleType()),
     ('Location',StringType())
]
schema = StructType([StructField(col[0], col[1], True) for col in labels])
schema<jupyter_output><empty_output><jupyter_text>-----  Open CSV and read the content using the format of customized Schema<jupyter_code>rc = spark.read.csv('../data/chicago_crimes.csv', schema=schema, header=True)
rc.printSchema()
rc.show(2)<jupyter_output>+--------+-----------+----+-------------------+----+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|      ID|Case Number|Date|              Block|IUCR|Primary Type|         Description|Location Description|Arrest|Domestic|Beat|District|Ward|Community Area|FBI Code|X Coordinate|Y Coordinate|Year|          Updated On|    Latitude|    Longitude|            Location|
+--------+-----------+----+-------------------+----+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|10224738|   HY411648|null|    043XX S WOOD ST|0486|     BATTERY|DOMESTIC BATTERY ...|           RESIDENCE| false|    true|0924|     009|  12|            61|     08B|     1165074|     1875917|2015|02[...]<jupyter_text>----------  2) Working with Columns<jupyter_code># get the file and preprocess it.
rc = spark.read.csv('../data/chicago_crimes.csv',header=True).withColumn('Date',to_timestamp(col('Date'),'MM/dd/yyyy hh:mm:ss a')).filter(col('Date') <= lit('2018-11-11'))<jupyter_output><empty_output><jupyter_text>Display only the first 5 rows of the column name IUCR<jupyter_code>rc.select('IUCR').show(5)
rc.select(rc.IUCR).show(5)
rc.select(col('IUCR')).show(5)<jupyter_output>+----+
|IUCR|
+----+
|0486|
|0870|
|0810|
|2023|
|0560|
+----+
only showing top 5 rows<jupyter_text>Display only the first 4 rows of column names Case Number, Date and Arrest<jupyter_code>rc.select('Case Number', 'Date', 'Arrest').show(4)<jupyter_output>+-----------+-------------------+------+
|Case Number|               Date|Arrest|
+-----------+-------------------+------+
|   HY411648|2015-09-05 13:30:00| false|
|   HY411615|2015-09-04 11:30:00| false|
|   JC213529|2018-09-01 00:01:00| false|
|   HY411595|2015-09-05 12:45:00|  true|
+-----------+-------------------+------+
only showing top 4 rows<jupyter_text>Add a column with name One, with entries of 1s- constant value column as a last column, to a table<jupyter_code>from pyspark.sql.functions import lit
# new column will be displayed as the last column
rc.withColumn('One', lit(1)).show(5)
rc.withColumn('Two', lit(2)).show(5)<jupyter_output>+--------+-----------+-------------------+--------------------+----+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+---+
|      ID|Case Number|               Date|               Block|IUCR|Primary Type|         Description|Location Description|Arrest|Domestic|Beat|District|Ward|Community Area|FBI Code|X Coordinate|Y Coordinate|Year|          Updated On|    Latitude|    Longitude|            Location|Two|
+--------+-----------+-------------------+--------------------+----+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+---+
|10224738|   HY411648|2015-09-05 13:30:00|     043XX S WOOD ST|0486|     BATTERY|DOMESTIC BATTERY ...|           RESIDENCE| false|    true|[...]<jupyter_text>Remove the Column IUCR<jupyter_code>temp = rc.drop('IUCR')
temp.show(2)<jupyter_output>+--------+-----------+-------------------+-------------------+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|      ID|Case Number|               Date|              Block|Primary Type|         Description|Location Description|Arrest|Domestic|Beat|District|Ward|Community Area|FBI Code|X Coordinate|Y Coordinate|Year|          Updated On|    Latitude|    Longitude|            Location|
+--------+-----------+-------------------+-------------------+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|10224738|   HY411648|2015-09-05 13:30:00|    043XX S WOOD ST|     BATTERY|DOMESTIC BATTERY ...|           RESIDENCE| false|    true|0924|     009|  12|            61|  [...]<jupyter_text>-------  3) Working with rows<jupyter_code># get the file and preprocess it.
rc = spark.read.csv('../data/chicago_crimes.csv',header=True).withColumn('Date',to_timestamp(col('Date'),'MM/dd/yyyy hh:mm:ss a')).filter(col('Date') <= lit('2018-11-11'))
rc.show(1)<jupyter_output>+--------+-----------+-------------------+---------------+----+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|      ID|Case Number|               Date|          Block|IUCR|Primary Type|         Description|Location Description|Arrest|Domestic|Beat|District|Ward|Community Area|FBI Code|X Coordinate|Y Coordinate|Year|          Updated On|    Latitude|    Longitude|            Location|
+--------+-----------+-------------------+---------------+----+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|10224738|   HY411648|2015-09-05 13:30:00|043XX S WOOD ST|0486|     BATTERY|DOMESTIC BATTERY ...|           RESIDENCE| false|    true|0924|     009|  12|            6[...]<jupyter_text>Add the reported crimes for an additional day, 12-Nov-2018 to our dataset.  first, filter data only for 12 Nov, 2018<jupyter_code># filter data only for 12 Nov, 2018
day_12_data = spark.read.csv('../data/chicago_crimes.csv',header=True).withColumn('Date',to_timestamp(col('Date'),'MM/dd/yyyy hh:mm:ss a')).filter(col('Date') == lit('2018-11-12'))
day_12_data.count()
day_12_data.show()<jupyter_output>+--------+-----------+-------------------+-------------------+----+------------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|      ID|Case Number|               Date|              Block|IUCR|      Primary Type|         Description|Location Description|Arrest|Domestic|Beat|District|Ward|Community Area|FBI Code|X Coordinate|Y Coordinate|Year|          Updated On|    Latitude|    Longitude|            Location|
+--------+-----------+-------------------+-------------------+----+------------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|11505149|   JB513151|2018-11-12 00:00:00| 003XX S WHIPPLE ST|0810|             THEFT|           OVER $500|              STREET| false| [...]<jupyter_text>union day 12 data to original df, check the last 5 rows<jupyter_code>rc.union(day_12_data).orderBy('Date', ascending=False).show(5)<jupyter_output>+--------+-----------+-------------------+-------------------+----+------------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|      ID|Case Number|               Date|              Block|IUCR|      Primary Type|         Description|Location Description|Arrest|Domestic|Beat|District|Ward|Community Area|FBI Code|X Coordinate|Y Coordinate|Year|          Updated On|    Latitude|    Longitude|            Location|
+--------+-----------+-------------------+-------------------+----+------------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|11505149|   JB513151|2018-11-12 00:00:00| 003XX S WHIPPLE ST|0810|             THEFT|           OVER $500|              STREET| false| [...]<jupyter_text>What are the top 10 number of reported crimes by Primary Type, in descending order occurence?<jupyter_code>rc.groupBy('Primary Type').count().show()
rc.groupBy('Primary Type').count().orderBy('count', ascending=False).show(10)<jupyter_output>+-------------------+-------+
|       Primary Type|  count|
+-------------------+-------+
|              THEFT|1418449|
|            BATTERY|1232233|
|    CRIMINAL DAMAGE| 771504|
|          NARCOTICS| 711744|
|      OTHER OFFENSE| 418860|
|            ASSAULT| 418509|
|           BURGLARY| 388033|
|MOTOR VEHICLE THEFT| 314131|
| DECEPTIVE PRACTICE| 266137|
|            ROBBERY| 255598|
+-------------------+-------+
only showing top 10 rows<jupyter_text>---------  4) Challenges- What percentage of reported crimes resulted in an arrest?- What are the top 3 locations for reported crimes?<jupyter_code># get the file and preprocess it.
rc = spark.read.csv('../data/chicago_crimes.csv',header=True).withColumn('Date',to_timestamp(col('Date'),'MM/dd/yyyy hh:mm:ss a')).filter(col('Date') <= lit('2018-11-12'))
rc.show(2)<jupyter_output>+--------+-----------+-------------------+-------------------+----+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|      ID|Case Number|               Date|              Block|IUCR|Primary Type|         Description|Location Description|Arrest|Domestic|Beat|District|Ward|Community Area|FBI Code|X Coordinate|Y Coordinate|Year|          Updated On|    Latitude|    Longitude|            Location|
+--------+-----------+-------------------+-------------------+----+------------+--------------------+--------------------+------+--------+----+--------+----+--------------+--------+------------+------------+----+--------------------+------------+-------------+--------------------+
|10224738|   HY411648|2015-09-05 13:30:00|    043XX S WOOD ST|0486|     BATTERY|DOMESTIC BATTERY ...|           RESIDENCE| false|    true|0924|     009|  [...]<jupyter_text>What percentage of reported crimes resulted in an arrest?<jupyter_code># find the total crimes with Arrest = True  /  Total Crime Cases
rc.select('Arrest').distinct().show()
rc.printSchema()
total_crimes = rc.count()
total_crimes
# crimes resulted in Arrest (arrest=True)
total_arrested_cirmes = rc.filter(col('Arrest') == 'true').count()
total_arrested_cirmes
# find the percentages
print("Percentage of reported crimes resulted in an arrest: {} %".format(100* total_arrested_cirmes / total_crimes))<jupyter_output>Percentage of reported crimes resulted in an arrest: 27.753353467288935 %<jupyter_text>------  What are the top 3 locations for reported crimes?<jupyter_code>rc.groupBy('Location Description').count().orderBy('count', ascending=False).show(3)<jupyter_output>+--------------------+-------+
|Location Description|  count|
+--------------------+-------+
|              STREET|1770732|
|           RESIDENCE|1145165|
|           APARTMENT| 698462|
+--------------------+-------+
only showing top 3 rows
{'fraction_non_alphanumeric': 0.38496355513755, 'fraction_numerical': 0.03867857982600517, 'mean_word_length': 1.4667246628969117, 'pattern_counts': {'":': 0, '<': 67, '<?xml version=': 0, '>': 63, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, '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': '19', 'dir': 'jupyter-structured-clean-dedup', 'id': '159120', 'license': 'MIT', 'n_tokens_mistral': 5024, 'n_tokens_neox': 4437, 'n_words': 903, 'path': 'Spark - Apache PySpark by Example/02.Working with DataFrame API/01.Working With Data Frame API.ipynb', 'repo_name': 'ptyadana/probability-and-statistics-for-business-and-data-science', 'size': '17012'}