Bigdata_Spark_Consultant

Tuesday, May 26, 2020

Spark Practice 2

>>> spark=SparkSession.builder.appName("Analyzing London Crime data").getOrCreate()
>>> data=spark.data.format("csv").option("header","true").load("/home/rameshkumar/datasets/london_crime_by_lsoa.csv")
Traceback (most recent call last):
File "", line 1, in
AttributeError: 'SparkSession' object has no attribute 'data'
>>> data=spark.read.format("csv").option("header","true").load("/home/rameshkumar/datasets/london_crime_by_lsoa.csv")
Traceback (most recent call last):
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 63, in deco
return f(*a, **kw)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py", line 320, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o59.load.
: org.apache.spark.sql.AnalysisException: Path does not exist: hdfs://nn01.itversity.com:8020/home/rameshkumar/datasets/london_crime_by_lsoa.csv;
at org.apache.spark.sql.execution.datasources.DataSource$.org$apache$spark$sql$execution$datasources$DataSource$$checkAndGlobPathIfNecessary(DataSource.scala:715)
at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$15.apply(DataSource.scala:389)
at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$15.apply(DataSource.scala:389)
at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
at scala.collection.immutable.List.foreach(List.scala:381)
at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
at scala.collection.immutable.List.flatMap(List.scala:344)
at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:388)
at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:239)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:174)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:748)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "", line 1, in
File "/usr/hdp/current/spark2-client/python/pyspark/sql/readwriter.py", line 166, in load
return self._df(self._jreader.load(path))
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py", line 1160, in __call__
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 69, in deco
raise AnalysisException(s.split(': ', 1)[1], stackTrace)
pyspark.sql.utils.AnalysisException: 'Path does not exist: hdfs://nn01.itversity.com:8020/home/rameshkumar/datasets/london_crime_by_lsoa.csv;'
>>> data=spark.read.format("csv").option("header","true").load("file:///home/rameshkumar/datasets/london_crime_by_lsoa.csv")
[Stage 0:> ( >>> data.printSchema()
root
|-- lsoa_code: string (nullable = true)
|-- borough: string (nullable = true)
|-- major_category: string (nullable = true)
|-- minor_category: string (nullable = true)
|-- value: string (nullable = true)
|-- year: string (nullable = true)
|-- month: string (nullable = true)

>>> data.count()
[Stage 1:> (0 + 16) / 16]
[Stage 1:======================================================================================================> (8[Stage 1:=========================================================================================================================================================> (12 13490604
>>>
>>> data.limit(5).show()
[Stage 3:> ( +---------+----------+--------------------+--------------------+-----+----+-----+
|lsoa_code| borough| major_category| minor_category|value|year|month|
+---------+----------+--------------------+--------------------+-----+----+-----+
|E01001116| Croydon| Burglary|Burglary in Other...| 0|2016| 11|
|E01001646| Greenwich|Violence Against ...| Other violence| 0|2016| 11|
|E01000677| Bromley|Violence Against ...| Other violence| 0|2015| 5|
|E01003774| Redbridge| Burglary|Burglary in Other...| 0|2016| 3|
|E01004563|Wandsworth| Robbery| Personal Property| 0|2008| 6|
+---------+----------+--------------------+--------------------+-----+----+-----+

>>> data.dropna()
DataFrame[lsoa_code: string, borough: string, major_category: string, minor_category: string, value: string, year: string, month: string]
>>> data.show(5)
[Stage 4:> ( +---------+----------+--------------------+--------------------+-----+----+-----+
|lsoa_code| borough| major_category| minor_category|value|year|month|
+---------+----------+--------------------+--------------------+-----+----+-----+
|E01001116| Croydon| Burglary|Burglary in Other...| 0|2016| 11|
|E01001646| Greenwich|Violence Against ...| Other violence| 0|2016| 11|
|E01000677| Bromley|Violence Against ...| Other violence| 0|2015| 5|
|E01003774| Redbridge| Burglary|Burglary in Other...| 0|2016| 3|
|E01004563|Wandsworth| Robbery| Personal Property| 0|2008| 6|
+---------+----------+--------------------+--------------------+-----+----+-----+
only showing top 5 rows

>>> data=data.drop('lsoa_code')
>>> data.show(5)
[Stage 5:> ( +----------+--------------------+--------------------+-----+----+-----+
| borough| major_category| minor_category|value|year|month|
+----------+--------------------+--------------------+-----+----+-----+
| Croydon| Burglary|Burglary in Other...| 0|2016| 11|
| Greenwich|Violence Against ...| Other violence| 0|2016| 11|
| Bromley|Violence Against ...| Other violence| 0|2015| 5|
| Redbridge| Burglary|Burglary in Other...| 0|2016| 3|
|Wandsworth| Robbery| Personal Property| 0|2008| 6|
+----------+--------------------+--------------------+-----+----+-----+
only showing top 5 rows

>>> total_boroughs=data.select('borough')
>>> total_boroughs.show()
[Stage 6:> ( +--------------------+
| borough|
+--------------------+
| Croydon|
| Greenwich|
| Bromley|
| Redbridge|
| Wandsworth|
| Ealing|
| Ealing|
| Hounslow|
| Newham|
| Sutton|
| Haringey|
| Lambeth|
|Richmond upon Thames|
| Wandsworth|
| Hillingdon|
| Havering|
|Barking and Dagenham|
| Redbridge|
|Kingston upon Thames|
| Sutton|
+--------------------+
only showing top 20 rows

>>> total_boroughs=data.select('borough').distinct()
>>> total_boroughs.show()
[Stage 7:> (0 [Stage 7:===================================================================================================================> (9[Stage 7:==================================================================================================================================================================================> (14[Stage 7:===============================================================================================================================================================================================> (15 +--------------------+
| borough|
+--------------------+
| Croydon|
| Wandsworth|
| Bexley|
| Lambeth|
|Barking and Dagenham|
| Camden|
| Greenwich|
| Newham|
| Tower Hamlets|
| Hounslow|
| Barnet|
| Harrow|
|Kensington and Ch...|
| Islington|
| Brent|
| Haringey|
| Bromley|
| Merton|
| Westminster|
| Hackney|
+--------------------+
only showing top 20 rows

>>> total_boroughs.count()
[Stage 15:> (0 [Stage 15:=========================> (2 [Stage 15:===============================================================> (5 [Stage 15:===========================================================================================================================================> (11[Stage 15:==============================================================================================================================================================================================> (15 33
>>> hackney_data=data.filter(data['borough']==='Hackney')
File "", line 1
hackney_data=data.filter(data['borough']==='Hackney')
^
SyntaxError: invalid syntax
>>> hackney_data=data.filter(data['borough']=='Hackney')
>>> hackney_data.show(5)
[Stage 18:> ( +-------+--------------------+--------------------+-----+----+-----+
|borough| major_category| minor_category|value|year|month|
+-------+--------------------+--------------------+-----+----+-----+
|Hackney| Criminal Damage|Criminal Damage T...| 0|2011| 6|
|Hackney|Violence Against ...| Harassment| 1|2013| 2|
|Hackney| Criminal Damage|Other Criminal Da...| 0|2011| 7|
|Hackney|Violence Against ...| Wounding/GBH| 0|2013| 12|
|Hackney| Theft and Handling| Other Theft Person| 0|2016| 8|
+-------+--------------------+--------------------+-----+----+-----+
only showing top 5 rows

>>> data_2015_2016=data.filter(data['year'].isin(['2015','2016']))
>>> data_2015_2016.sample(fraction=0.1).show()
[Stage 19:> ( +--------------------+--------------------+--------------------+-----+----+-----+
| borough| major_category| minor_category|value|year|month|
+--------------------+--------------------+--------------------+-----+----+-----+
| Croydon| Theft and Handling|Motor Vehicle Int...| 0|2016| 3|
| Waltham Forest| Burglary|Burglary in Other...| 0|2015| 6|
|Hammersmith and F...| Drugs| Possession Of Drugs| 1|2016| 2|
| Wandsworth| Theft and Handling|Handling Stolen G...| 0|2016| 2|
| Waltham Forest|Violence Against ...| Common Assault| 0|2016| 6|
|Barking and Dagenham| Criminal Damage|Criminal Damage T...| 0|2015| 4|
| Bexley| Robbery| Personal Property| 0|2016| 1|
| Croydon|Violence Against ...| Offensive Weapon| 0|2016| 2|
| Merton|Violence Against ...| Harassment| 1|2016| 6|
| Hackney| Theft and Handling| Other Theft Person| 0|2016| 8|
| Tower Hamlets| Theft and Handling| Other Theft| 2|2016| 1|
| Haringey|Violence Against ...| Harassment| 0|2016| 12|
| Westminster| Theft and Handling|Motor Vehicle Int...| 0|2016| 11|
| Lewisham| Drugs| Possession Of Drugs| 0|2015| 5|
| Haringey| Criminal Damage|Criminal Damage T...| 0|2015| 2|
| Lewisham| Fraud or Forgery|Other Fraud & For...| 0|2015| 6|
| Camden| Criminal Damage|Criminal Damage T...| 0|2015| 8|
| Southwark| Burglary|Burglary in Other...| 0|2016| 8|
| Lambeth| Theft and Handling| Other Theft| 9|2015| 5|
|Hammersmith and F...| Burglary|Burglary in Other...| 0|2016| 5|
+--------------------+--------------------+--------------------+-----+----+-----+
only showing top 20 rows

>>> data_2014_onwards=data.filter(data['year']>=2014)
>>> data_2014_onwards.sample(fraction=0.1).show()
[Stage 20:> ( +--------------------+--------------------+--------------------+-----+----+-----+
| borough| major_category| minor_category|value|year|month|
+--------------------+--------------------+--------------------+-----+----+-----+
| Bromley|Violence Against ...| Other violence| 0|2015| 5|
| Haringey|Violence Against ...| Wounding/GBH| 0|2015| 12|
| Haringey| Criminal Damage|Criminal Damage T...| 0|2016| 12|
| Ealing| Robbery| Personal Property| 0|2014| 9|
| Lambeth| Drugs| Drug Trafficking| 0|2015| 9|
| Newham|Violence Against ...| Harassment| 0|2014| 1|
| Lambeth| Sexual Offences| Rape| 0|2014| 3|
| Brent| Theft and Handling| Other Theft Person| 1|2016| 12|
| Ealing|Other Notifiable ...| Other Notifiable| 0|2016| 12|
| Barnet| Theft and Handling|Handling Stolen G...| 0|2016| 4|
| Havering|Other Notifiable ...| Other Notifiable| 0|2016| 6|
|Hammersmith and F...| Criminal Damage|Criminal Damage T...| 0|2015| 8|
| Islington| Burglary|Burglary in Other...| 1|2016| 7|
| Hillingdon|Violence Against ...| Wounding/GBH| 0|2016| 1|
| Lambeth| Drugs| Possession Of Drugs| 0|2014| 5|
| Greenwich| Criminal Damage|Criminal Damage T...| 0|2015| 1|
| Greenwich| Criminal Damage|Criminal Damage T...| 0|2016| 10|
| Hillingdon|Other Notifiable ...| Other Notifiable| 0|2015| 6|
| Wandsworth|Violence Against ...| Harassment| 0|2016| 11|
| Merton|Violence Against ...| Harassment| 1|2016| 6|
+--------------------+--------------------+--------------------+-----+----+-----+
only showing top 20 rows

>>> borough_crime_count=data.groupBy('borough').count()
>>> borough_crime_count.show(5)
[Stage 21:> (0 [Stage 21:============> (1 [Stage 21:=========================> (2 [Stage 21:==================================================> (4 [Stage 21:=========================================================================================> (7[Stage 21:==================================================================================================================> (9[Stage 21:====================================================================================================================================================================> (13[Stage 21:=================================================================================================================================================================================> (14[Stage 21:==============================================================================================================================================================================================> (15 +--------------------+------+
| borough| count|
+--------------------+------+
| Croydon|602100|
| Wandsworth|498636|
| Bexley|385668|
| Lambeth|519048|
|Barking and Dagenham|311040|
+--------------------+------+
only showing top 5 rows

>>> borough_conviction_sum=data.groupBy('borough').agg({"value":"sum"})
>>> borough_conviction_sum.show(5)
[Stage 29:> (0 [Stage 29:============> (1 [Stage 29:=========================> (2 [Stage 29:==================================================> (4 [Stage 29:============================================================================> (6 [Stage 29:======================================================================================================> (8[Stage 29:===========================================================================================================================================> (11 +--------------------+----------+
| borough|sum(value)|
+--------------------+----------+
| Croydon| 260294.0|
| Wandsworth| 204741.0|
| Bexley| 114136.0|
| Lambeth| 292178.0|
|Barking and Dagenham| 149447.0|
+--------------------+----------+
only showing top 5 rows

>>> total_borough_convictions=borough_conviction_sum.agg({"convictions":"sum"})
Traceback (most recent call last):
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 63, in deco
return f(*a, **kw)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py", line 320, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o135.agg.
: org.apache.spark.sql.AnalysisException: Cannot resolve column name "convictions" among (borough, sum(value));
at org.apache.spark.sql.Dataset$$anonfun$resolve$1.apply(Dataset.scala:222)
at org.apache.spark.sql.Dataset$$anonfun$resolve$1.apply(Dataset.scala:222)
at scala.Option.getOrElse(Option.scala:121)
at org.apache.spark.sql.Dataset.resolve(Dataset.scala:221)
at org.apache.spark.sql.Dataset.col(Dataset.scala:1241)
at org.apache.spark.sql.Dataset.apply(Dataset.scala:1208)
at org.apache.spark.sql.RelationalGroupedDataset$$anonfun$agg$2.apply(RelationalGroupedDataset.scala:172)
at org.apache.spark.sql.RelationalGroupedDataset$$anonfun$agg$2.apply(RelationalGroupedDataset.scala:171)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.immutable.Map$Map1.foreach(Map.scala:116)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at org.apache.spark.sql.RelationalGroupedDataset.agg(RelationalGroupedDataset.scala:171)
at org.apache.spark.sql.RelationalGroupedDataset.agg(RelationalGroupedDataset.scala:190)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:748)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "", line 1, in
File "/usr/hdp/current/spark2-client/python/pyspark/sql/dataframe.py", line 1327, in agg
return self.groupBy().agg(*exprs)
File "/usr/hdp/current/spark2-client/python/pyspark/sql/group.py", line 88, in agg
jdf = self._jgd.agg(exprs[0])
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py", line 1160, in __call__
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 69, in deco
raise AnalysisException(s.split(': ', 1)[1], stackTrace)
pyspark.sql.utils.AnalysisException: 'Cannot resolve column name "convictions" among (borough, sum(value));'
>>> total_borough_convictions = borough_conviction_sum.agg({"convictions":"sum"})
Traceback (most recent call last):
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 63, in deco
return f(*a, **kw)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py", line 320, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o167.agg.
: org.apache.spark.sql.AnalysisException: Cannot resolve column name "convictions" among (borough, sum(value));
at org.apache.spark.sql.Dataset$$anonfun$resolve$1.apply(Dataset.scala:222)
at org.apache.spark.sql.Dataset$$anonfun$resolve$1.apply(Dataset.scala:222)
at scala.Option.getOrElse(Option.scala:121)
at org.apache.spark.sql.Dataset.resolve(Dataset.scala:221)
at org.apache.spark.sql.Dataset.col(Dataset.scala:1241)
at org.apache.spark.sql.Dataset.apply(Dataset.scala:1208)
at org.apache.spark.sql.RelationalGroupedDataset$$anonfun$agg$2.apply(RelationalGroupedDataset.scala:172)
at org.apache.spark.sql.RelationalGroupedDataset$$anonfun$agg$2.apply(RelationalGroupedDataset.scala:171)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.immutable.Map$Map1.foreach(Map.scala:116)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at org.apache.spark.sql.RelationalGroupedDataset.agg(RelationalGroupedDataset.scala:171)
at org.apache.spark.sql.RelationalGroupedDataset.agg(RelationalGroupedDataset.scala:190)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:748)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "", line 1, in
File "/usr/hdp/current/spark2-client/python/pyspark/sql/dataframe.py", line 1327, in agg
return self.groupBy().agg(*exprs)
File "/usr/hdp/current/spark2-client/python/pyspark/sql/group.py", line 88, in agg
jdf = self._jgd.agg(exprs[0])
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py", line 1160, in __call__
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 69, in deco
raise AnalysisException(s.split(': ', 1)[1], stackTrace)
pyspark.sql.utils.AnalysisException: 'Cannot resolve column name "convictions" among (borough, sum(value));'
>>> borough_conviction_sum=data.groupBy('borough').agg({"value":"sum"}).withColumnRenamed("sum(value)","convictions")
>>> total_borough_convictions = borough_conviction_sum.agg({"convictions":"sum"})
>>> total_borough_convictions.show()
[Stage 37:> (0 [Stage 37:============> (1 [Stage 37:==================================================> (4 [Stage 37:============================================================================> (6 [Stage 37:======================================================================================================> (8[Stage 37:==============================================================================================================================================================================================> (15 +----------------+
|sum(convictions)|
+----------------+
| 6447758.0|
+----------------+

>>> total_convictions = total_borough_convictions.collect()
[Stage 40:> (0 [Stage 40:============> (1 [Stage 40:=========================================================================================> (7[Stage 40:==============================================================================================================================> (10[Stage 40:========================================================================================================================================================> (12[Stage 40:=================================================================================================================================================================================> (14[Stage 40:==============================================================================================================================================================================================> (15 >>> total_convictions = total_borough_convictions.collect()[0][0]
>>> total_convictions = total_borough_convictions.collect()
>>> total_convictions = total_borough_convictions.collect()[0][0]
>>> import pyspark.sql.functions as func
>>> borough_percentage_contribution = borough_conviction_sum.withColumn(\
... func.round(borough_conviction_sum.convictions / total_convictions * 100, 2))
Traceback (most recent call last):
File "", line 2, in
TypeError: withColumn() missing 1 required positional argument: 'col'
>>> borough_percentage_contribution = borough_conviction_sum.withColumn(
... "% contribution",
... func.round(borough_conviction_sum.convictions / total_convictions * 100, 2))
>>> borough_percentage_contribution.printSchema()
root
|-- borough: string (nullable = true)
|-- convictions: double (nullable = true)
|-- % contribution: double (nullable = true)

>>> borough_percentage_contribution.orderBy(borough_percentage_contribution[2].desc()).show()
[Stage 52:> (0 [Stage 52:============> (1 [Stage 52:=========================> (2 [Stage 52:==================================================> (4 [Stage 52:======================================================================================================> (8[Stage 52:====================================================================================================================================================================> (13 +--------------------+-----------+--------------+
| borough|convictions|% contribution|
+--------------------+-----------+--------------+
| Westminster| 455028.0| 7.06|
| Lambeth| 292178.0| 4.53|
| Southwark| 278809.0| 4.32|
| Camden| 275147.0| 4.27|
| Newham| 262024.0| 4.06|
| Croydon| 260294.0| 4.04|
| Ealing| 251562.0| 3.9|
| Islington| 230286.0| 3.57|
| Tower Hamlets| 228613.0| 3.55|
| Brent| 227551.0| 3.53|
| Hackney| 217119.0| 3.37|
| Lewisham| 215137.0| 3.34|
| Haringey| 213272.0| 3.31|
| Barnet| 212191.0| 3.29|
| Hillingdon| 209680.0| 3.25|
| Wandsworth| 204741.0| 3.18|
| Waltham Forest| 203879.0| 3.16|
| Enfield| 193880.0| 3.01|
| Hounslow| 186772.0| 2.9|
|Hammersmith and F...| 185259.0| 2.87|
+--------------------+-----------+--------------+
only showing top 20 rows

>>> conviction_monthly=data.filter(data['year']==2014).groupBy("month").agg({'value":"sum"}).withColumnRenamed("sum(value)","convictions")
File "", line 1
conviction_monthly=data.filter(data['year']==2014).groupBy("month").agg({'value":"sum"}).withColumnRenamed("sum(value)","convictions")
^
SyntaxError: EOL while scanning string literal
>>> conviction_monthly=data.filter(data['year']==2014).groupBy("month").agg({"value":"sum"}).withColumnRenamed("sum(value)","convictions")
>>> total_conviction_monthly = conviction_monthly.agg({"convictions":"sum"}) .collect()[0][0]
[Stage 54:> (0 [Stage 54:============> (1 >>> total_conviction_monthly = conviction_monthly.agg({"convictions":"sum"}) .collect()
[Stage 57:> (0 [Stage 57:============> (1 [Stage 57:======================================================================================================> (8[Stage 57:========================================================================================================================================================> (12[Stage 57:==============================================================================================================================================================================================> (15 >>> total_conviction_monthly = conviction_monthly .withColumn("percent",
... func.round(conviction_monthly.convictions/total_conviction_monthly * 100, 2))
Traceback (most recent call last):
File "", line 2, in
File "/usr/hdp/current/spark2-client/python/pyspark/sql/column.py", line 116, in _
njc = getattr(self._jc, name)(jc)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py", line 1160, in __call__
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 63, in deco
return f(*a, **kw)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py", line 320, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o247.divide.
: java.lang.RuntimeException: Unsupported literal type class java.util.ArrayList [[680183.0]]
at org.apache.spark.sql.catalyst.expressions.Literal$.apply(literals.scala:77)
at org.apache.spark.sql.catalyst.expressions.Literal$$anonfun$create$2.apply(literals.scala:163)
at org.apache.spark.sql.catalyst.expressions.Literal$$anonfun$create$2.apply(literals.scala:163)
at scala.util.Try.getOrElse(Try.scala:79)
at org.apache.spark.sql.catalyst.expressions.Literal$.create(literals.scala:162)
at org.apache.spark.sql.functions$.typedLit(functions.scala:113)
at org.apache.spark.sql.functions$.lit(functions.scala:96)
at org.apache.spark.sql.Column.$div(Column.scala:746)
at org.apache.spark.sql.Column.divide(Column.scala:761)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:748)

>>> total_conviction_monthly.columns
Traceback (most recent call last):
File "", line 1, in
AttributeError: 'list' object has no attribute 'columns'
>>> total_conviction_monthly = conviction_monthly .withColumn("percent",
... func.round(conviction_monthly.convictions/total_conviction_monthly * 100, 2))
Traceback (most recent call last):
File "", line 2, in
File "/usr/hdp/current/spark2-client/python/pyspark/sql/column.py", line 116, in _
njc = getattr(self._jc, name)(jc)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py", line 1160, in __call__
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 63, in deco
return f(*a, **kw)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py", line 320, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o272.divide.
: java.lang.RuntimeException: Unsupported literal type class java.util.ArrayList [[680183.0]]
at org.apache.spark.sql.catalyst.expressions.Literal$.apply(literals.scala:77)
at org.apache.spark.sql.catalyst.expressions.Literal$$anonfun$create$2.apply(literals.scala:163)
at org.apache.spark.sql.catalyst.expressions.Literal$$anonfun$create$2.apply(literals.scala:163)
at scala.util.Try.getOrElse(Try.scala:79)
at org.apache.spark.sql.catalyst.expressions.Literal$.create(literals.scala:162)
at org.apache.spark.sql.functions$.typedLit(functions.scala:113)
at org.apache.spark.sql.functions$.lit(functions.scala:96)
at org.apache.spark.sql.Column.$div(Column.scala:746)
at org.apache.spark.sql.Column.divide(Column.scala:761)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:748)

>>> total_conviction_monthly.columns
Traceback (most recent call last):
File "", line 1, in
AttributeError: 'list' object has no attribute 'columns'
>>>
>>>
>>> total_conviction_monthly = conviction_monthly .withColumn("percent",
... func.round(conviction_monthly.convictions/total_conviction_monthly * 100, 2))
Traceback (most recent call last):
File "", line 2, in
File "/usr/hdp/current/spark2-client/python/pyspark/sql/column.py", line 116, in _
njc = getattr(self._jc, name)(jc)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py", line 1160, in __call__
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 63, in deco
return f(*a, **kw)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py", line 320, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o297.divide.
: java.lang.RuntimeException: Unsupported literal type class java.util.ArrayList [[680183.0]]
at org.apache.spark.sql.catalyst.expressions.Literal$.apply(literals.scala:77)
at org.apache.spark.sql.catalyst.expressions.Literal$$anonfun$create$2.apply(literals.scala:163)
at org.apache.spark.sql.catalyst.expressions.Literal$$anonfun$create$2.apply(literals.scala:163)
at scala.util.Try.getOrElse(Try.scala:79)
at org.apache.spark.sql.catalyst.expressions.Literal$.create(literals.scala:162)
at org.apache.spark.sql.functions$.typedLit(functions.scala:113)
at org.apache.spark.sql.functions$.lit(functions.scala:96)
at org.apache.spark.sql.Column.$div(Column.scala:746)
at org.apache.spark.sql.Column.divide(Column.scala:761)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:748)

>>>
>>> crimes_category.orderBy(crimes_category.convictions.desc()).show()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'crimes_category' is not defined
>>> crimes_category=data.groupBy("major_category").agg({"value":"sum"}).withColumnRenamed("sum(value)","convictions")
>>> crimes_category.orderBy(crimes_category.convictions.desc()).show()
[Stage 60:> (0 [Stage 60:======================================> (3 [Stage 60:==================================================================================================================> (9[Stage 60:==============================================================================================================================> (10[Stage 60:===========================================================================================================================================> (11[Stage 60:==============================================================================================================================================================================================> (15 +--------------------+-----------+
| major_category|convictions|
+--------------------+-----------+
| Theft and Handling| 2661861.0|
|Violence Against ...| 1558081.0|
| Burglary| 754293.0|
| Criminal Damage| 630938.0|
| Drugs| 470765.0|
| Robbery| 258873.0|
|Other Notifiable ...| 106349.0|
| Fraud or Forgery| 5325.0|
| Sexual Offences| 1273.0|
+--------------------+-----------+

>>> year_df=data.select('year')
>>> year_df.agg({'year':'min'}).show()
[Stage 62:> (0 [Stage 62:======================================> (3 [Stage 62:======================================================================================================> (8[Stage 62:====================================================================================================================================================================> (13[Stage 62:==============================================================================================================================================================================================> (15 +---------+
|min(year)|
+---------+
| 2008|
+---------+

>>> year_df.agg({'year':'max'}).show()
[Stage 64:> (0 [Stage 64:============> (1 [Stage 64:============================================================================> (6 [Stage 64:==============================================================================================================================> (10[Stage 64:==============================================================================================================================================================================================> (15 +---------+
|max(year)|
+---------+
| 2016|
+---------+

>>> year_df.describe().show()
[Stage 66:> (0 [Stage 66:============> (1 [Stage 66:=========================> (2 [Stage 66:============================================================================> (6 [Stage 66:====================================================================================================================================================================> (13 +-------+------------------+
|summary| year|
+-------+------------------+
| count| 13490604|
| mean| 2012.0|
| stddev|2.5819889931674394|
| min| 2008|
| max| 2016|
+-------+------------------+

>>> data.crosstab('borough', 'major_category') .select('borough_major_category', 'Burglary', 'Drugs', 'Fraud or Forgery', 'Robbery') .show()
[Stage 68:> (0 [Stage 68:============> (1 [Stage 68:=========================================================================================> (7[Stage 68:====================================================================================================================================================================> (13 +----------------------+--------+-----+----------------+-------+
|borough_major_category|Burglary|Drugs|Fraud or Forgery|Robbery|
+----------------------+--------+-----+----------------+-------+
| Havering| 32400|32616| 5508| 27648|
| Merton| 26784|29160| 5724| 23652|
| Haringey| 31320|35424| 10368| 29484|
| Tower Hamlets| 31104|37368| 5400| 28512|
| Bromley| 42552|42336| 8532| 34668|
| Enfield| 39528|44064| 9720| 35532|
| Kingston upon Thames| 21168|22140| 3780| 15660|
| Westminster| 27648|32616| 8748| 25056|
| Richmond upon Thames| 24840|23004| 2808| 18468|
| Lewisham| 36504|43740| 11016| 34884|
| Brent| 37368|46980| 9288| 34128|
| Barking and Dagenham| 23760|26244| 7236| 22248|
| Redbridge| 34776|36504| 8532| 32400|
| Islington| 26568|34128| 5184| 26244|
| Hackney| 31104|38772| 7560| 29160|
| Newham| 35424|41580| 10692| 34452|
| City of London| 540| 756| 0| 540|
| Sutton| 26136|26784| 3024| 21060|
| Southwark| 35856|45144| 11772| 35316|
| Harrow| 29592|31212| 4752| 23976|
+----------------------+--------+-----+----------------+-------+
only showing top 20 rows

SPARK Practice 1

>>> from pyspark.sql.types import Row >>> from datetime imort datetime
File "", line 1
from datetime imort datetime
^
SyntaxError: invalid syntax
>>> from datetime import datetime
>>>
>>>
>>> simple_date=sc.parallelize([1,"Alice",50])
>>> simple_date=sc.parallelize(1,"Alice",50)
Traceback (most recent call last):
File "", line 1, in
TypeError: parallelize() takes from 2 to 3 positional arguments but 4 were given
>>> simple_date=sc.parallelize([1,"Alice",50])
>>> simple_date
ParallelCollectionRDD[1] at parallelize at PythonRDD.scala:175
>>> simple_data
Traceback (most recent call last):
File "", line 1, in
NameError: name 'simple_data' is not defined
>>> simple_data=sc.parallelize([1,"Alice",50])
>>> simple_data
ParallelCollectionRDD[2] at parallelize at PythonRDD.scala:175
>>> simple_date.count()
[Stage 0:> (0[Stage 0:> (0 3
>>> simple_data.count()
3
>>> simple_date.first()
1
>>> simple_date.first()
1
>>> simple_date.collect()
[1, 'Alice', 50]
>>> records=sc.parallelize([[1,"Alice",50],[2,"Bob",80]])
>>> records
ParallelCollectionRDD[11] at parallelize at PythonRDD.scala:175
>>> records.collect()
[[1, 'Alice', 50], [2, 'Bob', 80]]
>>> records.count()
2
>>> records.first()
[1, 'Alice', 50]
>>> records.take(2)
[[1, 'Alice', 50], [2, 'Bob', 80]]
>>> records.take(3)
[[1, 'Alice', 50], [2, 'Bob', 80]]
>>> records.take(0)
[]
>>> records.take()
Traceback (most recent call last):
File "", line 1, in
TypeError: take() missing 1 required positional argument: 'num'
>>> records.take(1)
[[1, 'Alice', 50]]
>>> records.collect()
[[1, 'Alice', 50], [2, 'Bob', 80]]
>>> df=records.toDF()
>>> df1=records.toDF("ID","Name","Age")
Traceback (most recent call last):
File "", line 1, in
TypeError: toDF() takes from 1 to 3 positional arguments but 4 were given
>>> df.show()
+---+-----+---+
| _1| _2| _3|
+---+-----+---+
| 1|Alice| 50|
| 2| Bob| 80|
+---+-----+---+

>>> df1=records.toDF("ID,Name",Age")
File "", line 1
df1=records.toDF("ID,Name",Age")
^
SyntaxError: EOL while scanning string literal
>>> df1=records.toDF("ID,Name,Age")
Traceback (most recent call last):
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 63, in deco
return f(*a, **kw)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py", line 320, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.sql.api.python.PythonSQLUtils.parseDataType.
: org.apache.spark.sql.catalyst.parser.ParseException:
mismatched input ',' expecting (line 1, pos 2)

== SQL ==
ID,Name,Age
--^^^

at org.apache.spark.sql.catalyst.parser.ParseException.withCommand(ParseDriver.scala:239)
at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parse(ParseDriver.scala:115)
at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parseDataType(ParseDriver.scala:39)
at org.apache.spark.sql.api.python.PythonSQLUtils$.parseDataType(PythonSQLUtils.scala:29)
at org.apache.spark.sql.api.python.PythonSQLUtils.parseDataType(PythonSQLUtils.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:748)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/usr/hdp/current/spark2-client/python/pyspark/sql/types.py", line 846, in _parse_datatype_string
return from_ddl_datatype(s)
File "/usr/hdp/current/spark2-client/python/pyspark/sql/types.py", line 838, in from_ddl_datatype
sc._jvm.org.apache.spark.sql.api.python.PythonSQLUtils.parseDataType(type_str).json())
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py", line 1160, in __call__
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 73, in deco
raise ParseException(s.split(': ', 1)[1], stackTrace)
pyspark.sql.utils.ParseException: "\nmismatched input ',' expecting (line 1, pos 2)\n\n== SQL ==\nID,Name,Age\n--^^^\n"

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 63, in deco
return f(*a, **kw)
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py", line 320, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.sql.api.python.PythonSQLUtils.parseDataType.
: org.apache.spark.sql.catalyst.parser.ParseException:
mismatched input ',' expecting ':'(line 1, pos 9)

== SQL ==
struct
---------^^^

at org.apache.spark.sql.catalyst.parser.ParseException.withCommand(ParseDriver.scala:239)
at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parse(ParseDriver.scala:115)
at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parseDataType(ParseDriver.scala:39)
at org.apache.spark.sql.api.python.PythonSQLUtils$.parseDataType(PythonSQLUtils.scala:29)
at org.apache.spark.sql.api.python.PythonSQLUtils.parseDataType(PythonSQLUtils.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:748)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/usr/hdp/current/spark2-client/python/pyspark/sql/types.py", line 850, in _parse_datatype_string
return from_ddl_datatype("struct<%s>" % s.strip())
File "/usr/hdp/current/spark2-client/python/pyspark/sql/types.py", line 838, in from_ddl_datatype
sc._jvm.org.apache.spark.sql.api.python.PythonSQLUtils.parseDataType(type_str).json())
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py", line 1160, in __call__
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 73, in deco
raise ParseException(s.split(': ', 1)[1], stackTrace)
pyspark.sql.utils.ParseException: "\nmismatched input ',' expecting ':'(line 1, pos 9)\n\n== SQL ==\nstruct\n---------^^^\n"

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "", line 1, in
File "/usr/hdp/current/spark2-client/python/pyspark/sql/session.py", line 58, in toDF
return sparkSession.createDataFrame(self, schema, sampleRatio)
File "/usr/hdp/current/spark2-client/python/pyspark/sql/session.py", line 638, in createDataFrame
schema = _parse_datatype_string(schema)
File "/usr/hdp/current/spark2-client/python/pyspark/sql/types.py", line 852, in _parse_datatype_string
raise e
File "/usr/hdp/current/spark2-client/python/pyspark/sql/types.py", line 842, in _parse_datatype_string
return from_ddl_schema(s)
File "/usr/hdp/current/spark2-client/python/pyspark/sql/types.py", line 834, in from_ddl_schema
sc._jvm.org.apache.spark.sql.types.StructType.fromDDL(type_str).json())
File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py", line 1160, in __call__
File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 73, in deco
raise ParseException(s.split(': ', 1)[1], stackTrace)
pyspark.sql.utils.ParseException: "\nextraneous input ',' expecting {'SELECT', 'FROM', 'ADD', 'AS', 'ALL', 'DISTINCT', 'WHERE', 'GROUP', 'BY', 'GROUPING', 'SETS', 'CUBE', 'ROLLUP', 'ORDER', 'HAVING', 'LIMIT', 'AT', 'OR', 'AND', 'IN', NOT, 'NO', 'EXISTS', 'BETWEEN', 'LIKE', RLIKE, 'IS', 'NULL', 'TRUE', 'FALSE', 'NULLS', 'ASC', 'DESC', 'FOR', 'INTERVAL', 'CASE', 'WHEN', 'THEN', 'ELSE', 'END', 'JOIN', 'CROSS', 'OUTER', 'INNER', 'LEFT', 'SEMI', 'RIGHT', 'FULL', 'NATURAL', 'ON', 'LATERAL', 'WINDOW', 'OVER', 'PARTITION', 'RANGE', 'ROWS', 'UNBOUNDED', 'PRECEDING', 'FOLLOWING', 'CURRENT', 'FIRST', 'AFTER', 'LAST', 'ROW', 'WITH', 'VALUES', 'CREATE', 'TABLE', 'DIRECTORY', 'VIEW', 'REPLACE', 'INSERT', 'DELETE', 'INTO', 'DESCRIBE', 'EXPLAIN', 'FORMAT', 'LOGICAL', 'CODEGEN', 'COST', 'CAST', 'SHOW', 'TABLES', 'COLUMNS', 'COLUMN', 'USE', 'PARTITIONS', 'FUNCTIONS', 'DROP', 'UNION', 'EXCEPT', 'MINUS', 'INTERSECT', 'TO', 'TABLESAMPLE', 'STRATIFY', 'ALTER', 'RENAME', 'ARRAY', 'MAP', 'STRUCT', 'COMMENT', 'SET', 'RESET', 'DATA', 'START', 'TRANSACTION', 'COMMIT', 'ROLLBACK', 'MACRO', 'IGNORE', 'BOTH', 'LEADING', 'TRAILING', 'IF', 'POSITION', 'DIV', 'PERCENT', 'BUCKET', 'OUT', 'OF', 'SORT', 'CLUSTER', 'DISTRIBUTE', 'OVERWRITE', 'TRANSFORM', 'REDUCE', 'SERDE', 'SERDEPROPERTIES', 'RECORDREADER', 'RECORDWRITER', 'DELIMITED', 'FIELDS', 'TERMINATED', 'COLLECTION', 'ITEMS', 'KEYS', 'ESCAPED', 'LINES', 'SEPARATED', 'FUNCTION', 'EXTENDED', 'REFRESH', 'CLEAR', 'CACHE', 'UNCACHE', 'LAZY', 'FORMATTED', 'GLOBAL', TEMPORARY, 'OPTIONS', 'UNSET', 'TBLPROPERTIES', 'DBPROPERTIES', 'BUCKETS', 'SKEWED', 'STORED', 'DIRECTORIES', 'LOCATION', 'EXCHANGE', 'ARCHIVE', 'UNARCHIVE', 'FILEFORMAT', 'TOUCH', 'COMPACT', 'CONCATENATE', 'CHANGE', 'CASCADE', 'RESTRICT', 'CLUSTERED', 'SORTED', 'PURGE', 'INPUTFORMAT', 'OUTPUTFORMAT', DATABASE, DATABASES, 'DFS', 'TRUNCATE', 'ANALYZE', 'COMPUTE', 'LIST', 'STATISTICS', 'PARTITIONED', 'EXTERNAL', 'DEFINED', 'REVOKE', 'GRANT', 'LOCK', 'UNLOCK', 'MSCK', 'REPAIR', 'RECOVER', 'EXPORT', 'IMPORT', 'LOAD', 'ROLE', 'ROLES', 'COMPACTIONS', 'PRINCIPALS', 'TRANSACTIONS', 'INDEX', 'INDEXES', 'LOCKS', 'OPTION', 'ANTI', 'LOCAL', 'INPATH', IDENTIFIER, BACKQUOTED_IDENTIFIER}(line 1, pos 2)\n\n== SQL ==\nID,Name,Age\n--^^^\n"
>>> df1=records.toDF(ID,Name,Age)
Traceback (most recent call last):
File "", line 1, in
NameError: name 'ID' is not defined
>>> df1=records.toDF(ID,Name,Age)
Traceback (most recent call last):
File "", line 1, in
NameError: name 'ID' is not defined
>>> df1=records.toDF('ID','Name','Age')
Traceback (most recent call last):
File "", line 1, in
TypeError: toDF() takes from 1 to 3 positional arguments but 4 were given
>>> records
ParallelCollectionRDD[11] at parallelize at PythonRDD.scala:175
>>> records.collect()
[[1, 'Alice', 50], [2, 'Bob', 80]]
>>> df1=records.toDF('ID','Name','Age')
Traceback (most recent call last):
File "", line 1, in
TypeError: toDF() takes from 1 to 3 positional arguments but 4 were given
>>> df.columnns
Traceback (most recent call last):
File "", line 1, in
File "/usr/hdp/current/spark2-client/python/pyspark/sql/dataframe.py", line 1182, in __getattr__
"'%s' object has no attribute '%s'" % (self.__class__.__name__, name))
AttributeError: 'DataFrame' object has no attribute 'columnns'
>>> records.columnds
Traceback (most recent call last):
File "", line 1, in
AttributeError: 'RDD' object has no attribute 'columnds'
>>> records.columns
Traceback (most recent call last):
File "", line 1, in
AttributeError: 'RDD' object has no attribute 'columns'
>>> df.columnns
Traceback (most recent call last):
File "", line 1, in
File "/usr/hdp/current/spark2-client/python/pyspark/sql/dataframe.py", line 1182, in __getattr__
"'%s' object has no attribute '%s'" % (self.__class__.__name__, name))
AttributeError: 'DataFrame' object has no attribute 'columnns'
>>> columns = ['ID','NAME_FIRST', 'DEPT_NAME']
>>> df1=records.toDF(*columns)
Traceback (most recent call last):
File "", line 1, in
TypeError: toDF() takes from 1 to 3 positional arguments but 4 were given
>>> columns = ['ID','NAME_FIRST', 'DEPT_NAME']
>>> df2 = records.toDF(columns)
>>> df.show()
+---+-----+---+
| _1| _2| _3|
+---+-----+---+
| 1|Alice| 50|
| 2| Bob| 80|
+---+-----+---+

>>> df2.show()
+---+----------+---------+
| ID|NAME_FIRST|DEPT_NAME|
+---+----------+---------+
| 1| Alice| 50|
| 2| Bob| 80|
+---+----------+---------+

>>> data=sc.parallelize([Row(id=1,name="Alice",score=50)])
>>> data.count()
1
>>> data.collect()
[Row(id=1, name='Alice', score=50)]
>>> df=data.toDF()
>>> df.show()
+---+-----+-----+
| id| name|score|
+---+-----+-----+
| 1|Alice| 50|
+---+-----+-----+

>>> data=sc.parallelize([Row(id=1,name="Alice",score=50),\
... Row(id=2,name="Bob",score=80),\
... Row(id=3,name="Charlee",score=75)])
>>> df.data.toDF()
Traceback (most recent call last):
File "", line 1, in
File "/usr/hdp/current/spark2-client/python/pyspark/sql/dataframe.py", line 1182, in __getattr__
"'%s' object has no attribute '%s'" % (self.__class__.__name__, name))
AttributeError: 'DataFrame' object has no attribute 'data'
>>> df=data.toDF()
>>> df.show()
+---+-------+-----+
| id| name|score|
+---+-------+-----+
| 1| Alice| 50|
| 2| Bob| 80|
| 3|Charlee| 75|
+---+-------+-----+

>>> complex_data=sc.parallelize([Row(col_float=1.44,col_integer=10,col_string="John")])
>>> complex_data_df=complex_data.toDF()
>>> complex_Data_df.show()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'complex_Data_df' is not defined
>>> complex_data_df.show()
+---------+-----------+----------+
|col_float|col_integer|col_string|
+---------+-----------+----------+
| 1.44| 10| John|
+---------+-----------+----------+

>>> complex_data=sc.parallelize([Row(col_float=1.44,\
... col_integer=10,\
... col_string="John",\
... col_boolean=True,\
... col_list=[1,2,3])
...
...
...
... \
... ])
>>> complex_data_df=complex_Data.toDF()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'complex_Data' is not defined
>>> complex_data_df=complex_data.toDF()
>>> complex_Data_df.show()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'complex_Data_df' is not defined
>>> complex_data_df.show()
+-----------+---------+-----------+---------+----------+
|col_boolean|col_float|col_integer| col_list|col_string|
+-----------+---------+-----------+---------+----------+
| true| 1.44| 10|[1, 2, 3]| John|
+-----------+---------+-----------+---------+----------+

>>> complex_data=sc.parallelize([Row(\
... col_list=[1,2,3],\
... col_dict={"k1":0,"k2":1,"k3":2},\
... col_row=Row(columnA=10,columnB=20,columnC=30),\
... col_time=datetime(2014,8,1,14,1,5) \
... )])
>>> complex_data_df=complex_Data.toDF()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'complex_Data' is not defined
>>> complex_data_df=complex_data.toDF()
>>> complex_data_df.show()
+--------------------+---------+------------+-------------------+
| col_dict| col_list| col_row| col_time|
+--------------------+---------+------------+-------------------+
|[k3 -> 2, k1 -> 0...|[1, 2, 3]|[10, 20, 30]|2014-08-01 14:01:05|
+--------------------+---------+------------+-------------------+

>>> complex_data = sc.parallelize([Row(
... col_list = [1, 2, 3],
... col_dict = {"k1": 0},
... col_row = Row(a=10, b=20, c=30),
... col_time = datetime(2014, 8, 1, 14, 1, 5)
... ),
... Row(
... col_list = [1, 2, 3, 4, 5],
... col_dict = {"k1": 0,"k2": 1 },
... col_row = Row(a=40, b=50, c=60),
... col_time = datetime(2014, 8, 2, 14, 1, 6)
... ),
... Row(
... col_list = [1, 2, 3, 4, 5, 6, 7],
... col_dict = {"k1": 0, "k2": 1, "k3": 2 },
... col_row = Row(a=70, b=80, c=90),
... col_time = datetime(2014, 8, 3, 14, 1, 7)
... )])
>>>
>>> complex_data_df = complex_data.toDF()
>>> complex_data_df.show()
+--------------------+--------------------+------------+-------------------+
| col_dict| col_list| col_row| col_time|
+--------------------+--------------------+------------+-------------------+
| [k1 -> 0]| [1, 2, 3]|[10, 20, 30]|2014-08-01 14:01:05|
| [k1 -> 0, k2 -> 1]| [1, 2, 3, 4, 5]|[40, 50, 60]|2014-08-02 14:01:06|
|[k3 -> 2, k1 -> 0...|[1, 2, 3, 4, 5, 6...|[70, 80, 90]|2014-08-03 14:01:07|
+--------------------+--------------------+------------+-------------------+

>>> sqlContext=SQLContext(sc)
>>> sqlContext

>>> df=sqlContext.range(5)
>>> df
DataFrame[id: bigint]
>>> df.printSchema

>>> df.printSchema()
root
|-- id: long (nullable = false)

>>> df.show()
+---+
| id|
+---+
| 0|
| 1|
| 2|
| 3|
| 4|
+---+

>>> df.count()
5
>>> data=[('Alice',50),('Bob',80),('Charlee',75)]
>>> sqlContext.createDataframe(data).show()
Traceback (most recent call last):
File "", line 1, in
AttributeError: 'SQLContext' object has no attribute 'createDataframe'
>>> sqlContext.createDataFrame(data).show()
+-------+---+
| _1| _2|
+-------+---+
| Alice| 50|
| Bob| 80|
|Charlee| 75|
+-------+---+

>>> sqlContext.createDataFrame(data,['Name','Score']).show()
+-------+-----+
| Name|Score|
+-------+-----+
| Alice| 50|
| Bob| 80|
|Charlee| 75|
+-------+-----+

>>> complex_data = [
... (1.0,
... 10,
... "Alice",
... True,
... [1, 2, 3],
... {"k1": 0},
... Row(a=1, b=2, c=3),
... datetime(2014, 8, 1, 14, 1, 5)),
...
... (2.0,
... 20,
... "Bob",
... True,
... [1, 2, 3, 4, 5],
... {"k1": 0,"k2": 1 },
... Row(a=1, b=2, c=3),
... datetime(2014, 8, 1, 14, 1, 5)),
...
... (3.0,
... 30,
... "Charlee",
... False,
... [1, 2, 3, 4, 5, 6],
... {"k1": 0, "k2": 1, "k3": 2 },
... Row(a=1, b=2, c=3),
... datetime(2014, 8, 1, 14, 1, 5))
... ]
>>> sqlContext.createDataFrame(complex_Data).show()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'complex_Data' is not defined
>>> sqlContext.createDataFrame(complex_data).show()
+---+---+-------+-----+------------------+--------------------+---------+-------------------+
| _1| _2| _3| _4| _5| _6| _7| _8|
+---+---+-------+-----+------------------+--------------------+---------+-------------------+
|1.0| 10| Alice| true| [1, 2, 3]| [k1 -> 0]|[1, 2, 3]|2014-08-01 14:01:05|
|2.0| 20| Bob| true| [1, 2, 3, 4, 5]| [k1 -> 0, k2 -> 1]|[1, 2, 3]|2014-08-01 14:01:05|
|3.0| 30|Charlee|false|[1, 2, 3, 4, 5, 6]|[k3 -> 2, k1 -> 0...|[1, 2, 3]|2014-08-01 14:01:05|
+---+---+-------+-----+------------------+--------------------+---------+-------------------+

>>> complex_data_df = sqlContext.createDataFrame(complex_data, [
... 'col_integer',
... 'col_float',
... 'col_string',
... 'col_boolean',
... 'col_list',
... 'col_dictionary',
... 'col_row',
... 'col_date_time']
... )
>>> complex_data_df.show()
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|col_integer|col_float|col_string|col_boolean| col_list| col_dictionary| col_row| col_date_time|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
| 1.0| 10| Alice| true| [1, 2, 3]| [k1 -> 0]|[1, 2, 3]|2014-08-01 14:01:05|
| 2.0| 20| Bob| true| [1, 2, 3, 4, 5]| [k1 -> 0, k2 -> 1]|[1, 2, 3]|2014-08-01 14:01:05|
| 3.0| 30| Charlee| false|[1, 2, 3, 4, 5, 6]|[k3 -> 2, k1 -> 0...|[1, 2, 3]|2014-08-01 14:01:05|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+

>>>
>>> data=sc.parallelize([Row(1,"Alice",50),\
... Row(2,"Bob",80),\
... Row(3,"Charlee",75)])
>>> column_names=Row('id','name','score')
>>> students=data.map(lambda r:column_names(*r))
>>> students
PythonRDD[173] at RDD at PythonRDD.scala:48
>>> students.collect()
[Row(id=1, name='Alice', score=50), Row(id=2, name='Bob', score=80), Row(id=3, name='Charlee', score=75)]
>>> students_df=sqlContext.createDataFrame(students)
>>> students_df;
DataFrame[id: bigint, name: string, score: bigint]
>>> students_df.show()
+---+-------+-----+
| id| name|score|
+---+-------+-----+
| 1| Alice| 50|
| 2| Bob| 80|
| 3|Charlee| 75|
+---+-------+-----+

>>> complex_data_df.first()
Row(col_integer=1.0, col_float=10, col_string='Alice', col_boolean=True, col_list=[1, 2, 3], col_dictionary={'k1': 0}, col_row=Row(a=1, b=2, c=3), col_date_time=datetime.datetime(2014, 8, 1, 14, 1, 5))
>>> complex_data_df.first()
Row(col_integer=1.0, col_float=10, col_string='Alice', col_boolean=True, col_list=[1, 2, 3], col_dictionary={'k1': 0}, col_row=Row(a=1, b=2, c=3), col_date_time=datetime.datetime(2014, 8, 1, 14, 1, 5))
>>> complex_data_df.take(2)
[Row(col_integer=1.0, col_float=10, col_string='Alice', col_boolean=True, col_list=[1, 2, 3], col_dictionary={'k1': 0}, col_row=Row(a=1, b=2, c=3), col_date_time=datetime.datetime(2014, 8, 1, 14, 1, 5)), Row(col_integer=2.0, col_float=20, col_string='Bob', col_boolean=True, col_list=[1, 2, 3, 4, 5], col_dictionary={'k1': 0, 'k2': 1}, col_row=Row(a=1, b=2, c=3), col_date_time=datetime.datetime(2014, 8, 1, 14, 1, 5))]
>>> cell_list=complex_Data_df.collect()[0[2]
... cell_list=complex_Data_df.collect()[0[2]
File "", line 2
cell_list=complex_Data_df.collect()[0[2]
^
SyntaxError: invalid syntax
>>> cell_list=complex_data_df.collect()[0[2]
...
...
... }
File "", line 4
}
^
SyntaxError: invalid syntax
>>> cell_list=complex_data_df.collect()[0][2]
>>> cell_list
'Alice'
>>> cell_list=complex_data_Df.collect()[0][4]
Traceback (most recent call last):
File "", line 1, in
NameError: name 'complex_data_Df' is not defined
>>> cell_list=complex_data_df.collect()[0][4]
>>> cell_list
[1, 2, 3]
>>> cell_list.append(100)
>>> cell_list
[1, 2, 3, 100]
>>> complex_data_df.show()
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|col_integer|col_float|col_string|col_boolean| col_list| col_dictionary| col_row| col_date_time|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
| 1.0| 10| Alice| true| [1, 2, 3]| [k1 -> 0]|[1, 2, 3]|2014-08-01 14:01:05|
| 2.0| 20| Bob| true| [1, 2, 3, 4, 5]| [k1 -> 0, k2 -> 1]|[1, 2, 3]|2014-08-01 14:01:05|
| 3.0| 30| Charlee| false|[1, 2, 3, 4, 5, 6]|[k3 -> 2, k1 -> 0...|[1, 2, 3]|2014-08-01 14:01:05|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+

>>> complex_data_df.rdd.map(lambda x:(x.col_string,x.col_dictionary)).collect()
[('Alice', {'k1': 0}), ('Bob', {'k1': 0, 'k2': 1}), ('Charlee', {'k3': 2, 'k1': 0, 'k2': 1})]
>>> complex_data_df.select('col_string','col_list','col_date_time').show()
+----------+------------------+-------------------+
|col_string| col_list| col_date_time|
+----------+------------------+-------------------+
| Alice| [1, 2, 3]|2014-08-01 14:01:05|
| Bob| [1, 2, 3, 4, 5]|2014-08-01 14:01:05|
| Charlee|[1, 2, 3, 4, 5, 6]|2014-08-01 14:01:05|
+----------+------------------+-------------------+

>>> complex_data_df.rdd.map(lambda x:(x.col_string+"Boo")).collect()
['AliceBoo', 'BobBoo', 'CharleeBoo']
>>> complex_data_df.select('col_integer','col_float').\
... withColum("col_sum",complex_data_df.col_integer + complex_data_df.col_float).show()
Traceback (most recent call last):
File "", line 1, in
File "/usr/hdp/current/spark2-client/python/pyspark/sql/dataframe.py", line 1182, in __getattr__
"'%s' object has no attribute '%s'" % (self.__class__.__name__, name))
AttributeError: 'DataFrame' object has no attribute 'withColum'
>>> complex_data_df.select('col_integer','col_float').\
... withColumn("col_sum",complex_data_df.col_integer + complex_data_df.col_float).show()
+-----------+---------+-------+
|col_integer|col_float|col_sum|
+-----------+---------+-------+
| 1.0| 10| 11.0|
| 2.0| 20| 22.0|
| 3.0| 30| 33.0|
+-----------+---------+-------+

>>> complex_data_df.select('col_integer','col_float').\
... withColumn("col_sum",col_integer +col_float).show()
Traceback (most recent call last):
File "", line 2, in
NameError: name 'col_integer' is not defined
>>> complex_Data_df.select('col_boolean').withColumn("col_opposite",complex_data_df.col_boolean == False ).show()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'complex_Data_df' is not defined
>>> complex_data_df.select('col_boolean').withColumn("col_opposite",complex_data_df.col_boolean == False ).show()
+-----------+------------+
|col_boolean|col_opposite|
+-----------+------------+
| true| false|
| true| false|
| false| true|
+-----------+------------+

>>> complext_data_df.withColumnRenamed("col_dictionary","col_map").show()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'complext_data_df' is not defined
>>> complex_data_df.withColumnRenamed("col_dictionary","col_map").show()
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|col_integer|col_float|col_string|col_boolean| col_list| col_map| col_row| col_date_time|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
| 1.0| 10| Alice| true| [1, 2, 3]| [k1 -> 0]|[1, 2, 3]|2014-08-01 14:01:05|
| 2.0| 20| Bob| true| [1, 2, 3, 4, 5]| [k1 -> 0, k2 -> 1]|[1, 2, 3]|2014-08-01 14:01:05|
| 3.0| 30| Charlee| false|[1, 2, 3, 4, 5, 6]|[k3 -> 2, k1 -> 0...|[1, 2, 3]|2014-08-01 14:01:05|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+

>>> complext_data_df.withColumnRenamed(col_dictionary,col_map).show() Traceback (most recent call last):
File "", line 1, in
NameError: name 'complext_data_df' is not defined
>>> complext_data_data.select(col_dictionary).show()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'complext_data_data' is not defined
>>> complex_data_df.withColumnRenamed(col_dictionary,col_map).show()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'col_dictionary' is not defined
>>> complex_data_data.select(col_dictionary).show()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'complex_data_data' is not defined
>>> complex_data_data.select(complex_data_dfcol_dictionary).show()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'complex_data_data' is not defined
>>> complex_data_df.select(col_dictionary).show()
Traceback (most recent call last):
File "", line 1, in
NameError: name 'col_dictionary' is not defined
>>> complex_data_df.select(complex_data_df.col_dictionary).show()
+--------------------+
| col_dictionary|
+--------------------+
| [k1 -> 0]|
| [k1 -> 0, k2 -> 1]|
|[k3 -> 2, k1 -> 0...|
+--------------------+

>>> complex_data_df.select(complex_data_df.col_dictionary.alias("Name").show()
... )
Traceback (most recent call last):
File "", line 1, in
TypeError: 'Column' object is not callable
>>> complex_data_df.select(complex_data_df.col_string.alias("Name")).show()
+-------+
| Name|
+-------+
| Alice|
| Bob|
|Charlee|
+-------+

>>> import pandas
>>> df_pandas=complex_data_df.toPandas()
>>> df_pandas
col_integer col_float col_string col_boolean col_list col_dictionary col_row col_date_time
0 1.0 10 Alice True [1, 2, 3] {'k1': 0} (1, 2, 3) 2014-08-01 14:01:05
1 2.0 20 Bob True [1, 2, 3, 4, 5] {'k1': 0, 'k2': 1} (1, 2, 3) 2014-08-01 14:01:05
2 3.0 30 Charlee False [1, 2, 3, 4, 5, 6] {'k3': 2, 'k1': 0, 'k2': 1} (1, 2, 3) 2014-08-01 14:01:05
>>> df_spark=sqlContext.createDataFrame(df_pandas).show()
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|col_integer|col_float|col_string|col_boolean| col_list| col_dictionary| col_row| col_date_time|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
| 1.0| 10| Alice| true| [1, 2, 3]| [k1 -> 0]|[1, 2, 3]|2014-08-01 14:01:05|
| 2.0| 20| Bob| true| [1, 2, 3, 4, 5]| [k1 -> 0, k2 -> 1]|[1, 2, 3]|2014-08-01 14:01:05|
| 3.0| 30| Charlee| false|[1, 2, 3, 4, 5, 6]|[k3 -> 2, k1 -> 0...|[1, 2, 3]|2014-08-01 14:01:05|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+

Bigdata_Spark_Consultant

Tuesday, May 26, 2020

Spark Practice 2

SPARK Practice 1

Python Challenges Program

Blog Archive

Labels