While running K means clustering using pyspark, I am using the following lines of code to find the optimal K value. But there is some error constantly popping up at the model fitting line.
The preprocessing stages included removing NAs and label encoding,
from pyspark.sql.functions import when,col
#Encode column "Potential" from dataframe df
high = list(range(86,101))
middle = list(range (71, 86))
low = list(range(56, 71))
very_low = list(range(45,56))
#checking for NULL
df = df.where(col("Potential").isNotNull())
#encoding the column potential
df1 = df.withColumn("Potential_Grading", when(col("Potential").isin(high), "high").
when(col("Potential").isin(middle), "Middle").
when(col("Potential").isin(low), "Low").
when(col("Potential").isin(very_low),"Verylow"))
df_kmeans = df1.select("ID","Height(CM)","Weight(KG)", "Crossing", "Finishing", "HeadingAccuracy", "ShortPassing",
"Volleys", "Dribbling", "Curve", "FKAccuracy", "LongPassing", "BallControl", "Acceleration",
"SprintSpeed", "Agility", "Reactions", "Balance", "ShotPower", "Jumping", "Stamina", "Strength",
"LongShots", "Aggression", "Interceptions", "Vision", "Penalties", "Composure", "Positioning",
"Marking", "StandingTackle", "SlidingTackle","Potential_Grading")
FEATURES_COL1 = ['Height(CM)', 'Weight(KG)',
'Crossing', 'Finishing', 'HeadingAccuracy',
'ShortPassing', 'Volleys', 'Dribbling', 'Curve',
'FKAccuracy', 'LongPassing', 'BallControl',
'Acceleration', 'SprintSpeed', 'Agility',
'Reactions', 'Balance', 'ShotPower', 'Jumping',
'Stamina', 'Strength', 'LongShots', 'Aggression',
'Interceptions', 'Vision', 'Penalties','Positioning',
'Composure', 'Marking', 'StandingTackle', 'SlidingTackle']
from pyspark.ml.clustering import KMeans
from pyspark.ml.feature import VectorAssembler
#Vector assembler is a transformer which takes all of the columns specified under FEATURES_COL and combines them into a new vector column.
vecAssembler = VectorAssembler(inputCols=FEATURES_COL1, outputCol="features")
#df_kmeans_1 = vecAssembler.transform(df_kmeans).select('ID','features')
df_kmeans_1 = vecAssembler.transform(df_kmeans).select('ID','features')
The lines of code for fitting K means
cost = np.zeros(21)
from pyspark.ml.clustering import KMeans
for k in range(2,21):
# Trains a k-means model.
kmeans = KMeans().setK(k).setFeaturesCol("features").setSeed(1)
model1 = kmeans.fit(df_kmeans_1)
#Evaluate clustering by computing Within Set Sum of Squared Errors.
cost[k] = model1.computeCost(df_kmeans_1)
I am getting an error at the model fitting line of code:
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-15-2840daf75485> in <module>()
8 # Trains a k-means model.
9 kmeans = KMeans().setK(k).setFeaturesCol("features").setSeed(1)
---> 10 model1 = kmeans.fit(df_kmeans_1)
11
12 #Evaluate clustering by computing Within Set Sum of Squared Errors.
5 frames
/content/spark-2.4.0-bin-hadoop2.7/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
326 raise Py4JJavaError(
327 "An error occurred while calling {0}{1}{2}.\n".
--> 328 format(target_id, ".", name), value)
329 else:
330 raise Py4JError(
Py4JJavaError: An error occurred while calling o346.fit.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 15.0 failed 1 times, most recent failure: Lost task 0.0 in stage 15.0 (TID 18, localhost, executor driver): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$4: (struct<Height(CM):double,Weight(KG):double,Crossing_double_VectorAssembler_f596de6f26ae:double,Finishing_double_VectorAssembler_f596de6f26ae:double,HeadingAccuracy_double_VectorAssembler_f596de6f26ae:double,ShortPassing_double_VectorAssembler_f596de6f26ae:double,Volleys_double_VectorAssembler_f596de6f26ae:double,Dribbling_double_VectorAssembler_f596de6f26ae:double,Curve_double_VectorAssembler_f596de6f26ae:double,FKAccuracy_double_VectorAssembler_f596de6f26ae:double,LongPassing_double_VectorAssembler_f596de6f26ae:double,BallControl_double_VectorAssembler_f596de6f26ae:double,Acceleration_double_VectorAssembler_f596de6f26ae:double,SprintSpeed_double_VectorAssembler_f596de6f26ae:double,Agility_double_VectorAssembler_f596de6f26ae:double,Reactions_double_VectorAssembler_f596de6f26ae:double,Balance_double_VectorAssembler_f596de6f26ae:double,ShotPower_double_VectorAssembler_f596de6f26ae:double,Jumping_double_VectorAssembler_f596de6f26ae:double,Stamina_double_VectorAssembler_f596de6f26ae:double,Strength_double_VectorAssembler_f596de6f26ae:double,LongShots_double_VectorAssembler_f596de6f26ae:double,Aggression_double_VectorAssembler_f596de6f26ae:double,Interceptions_double_VectorAssembler_f596de6f26ae:double,V...
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$11$$anon$1.hasNext(WholeStageCodegenExec.scala:619)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:220)
at org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:298)
at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1165)
at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1156)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1091)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1156)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:882)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:335)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:286)
at org.apache.spark.rdd.ZippedPartitionsRDD2.compute(ZippedPartitionsRDD.scala:89)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:121)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.spark.SparkException: Encountered null while assembling a row with handleInvalid = "keep". Consider
removing nulls from dataset or using handleInvalid = "keep" or "skip".
at org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler.scala:287)
at org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler.scala:255)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35)
at org.apache.spark.ml.feature.VectorAssembler$.assemble(VectorAssembler.scala:255)
at org.apache.spark.ml.feature.VectorAssembler$$anonfun$4.apply(VectorAssembler.scala:144)
at org.apache.spark.ml.feature.VectorAssembler$$anonfun$4.apply(VectorAssembler.scala:143)
... 29 more
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1887)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1875)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1874)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1874)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:926)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:926)
at scala.Option.foreach(Option.scala:257)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:926)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2108)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2057)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2046)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:737)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2061)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2082)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2101)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2126)
at org.apache.spark.rdd.RDD.count(RDD.scala:1168)
at org.apache.spark.rdd.RDD$$anonfun$takeSample$1.apply(RDD.scala:572)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
at org.apache.spark.rdd.RDD.takeSample(RDD.scala:561)
at org.apache.spark.mllib.clustering.KMeans.initKMeansParallel(KMeans.scala:386)
at org.apache.spark.mllib.clustering.KMeans.runAlgorithm(KMeans.scala:282)
at org.apache.spark.mllib.clustering.KMeans.run(KMeans.scala:251)
at org.apache.spark.ml.clustering.KMeans$$anonfun$fit$1.apply(KMeans.scala:362)
at org.apache.spark.ml.clustering.KMeans$$anonfun$fit$1.apply(KMeans.scala:340)
at org.apache.spark.ml.util.Instrumentation$$anonfun$11.apply(Instrumentation.scala:183)
at scala.util.Try$.apply(Try.scala:192)
at org.apache.spark.ml.util.Instrumentation$.instrumented(Instrumentation.scala:183)
at org.apache.spark.ml.clustering.KMeans.fit(KMeans.scala:340)
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:238)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.spark.SparkException: Failed to execute user defined function($anonfun$4: (struct<Height(CM):double,Weight(KG):double,Crossing_double_VectorAssembler_f596de6f26ae:double,Finishing_double_VectorAssembler_f596de6f26ae:double,HeadingAccuracy_double_VectorAssembler_f596de6f26ae:double,ShortPassing_double_VectorAssembler_f596de6f26ae:double,Volleys_double_VectorAssembler_f596de6f26ae:double,Dribbling_double_VectorAssembler_f596de6f26ae:double,Curve_double_VectorAssembler_f596de6f26ae:double,FKAccuracy_double_VectorAssembler_f596de6f26ae:double,LongPassing_double_VectorAssembler_f596de6f26ae:double,BallControl_double_VectorAssembler_f596de6f26ae:double,Acceleration_double_VectorAssembler_f596de6f26ae:double,SprintSpeed_double_VectorAssembler_f596de6f26ae:double,Agility_double_VectorAssembler_f596de6f26ae:double,Reactions_double_VectorAssembler_f596de6f26ae:double,Balance_double_VectorAssembler_f596de6f26ae:double,ShotPower_double_VectorAssembler_f596de6f26ae:double,Jumping_double_VectorAssembler_f596de6f26ae:double,Stamina_double_VectorAssembler_f596de6f26ae:double,Strength_double_VectorAssembler_f596de6f26ae:double,LongShots_double_VectorAssembler_f596de6f26ae:double,Aggression_double_VectorAssembler_f596de6f26ae:double,Interceptions_double_VectorAssembler_f596de6f26ae:double,Vision_double_VectorAssembler_f596de6f26ae:double,Penalties_double_VectorAssembler_f596de6f26ae:double,Positioning_double_VectorAssembler_f596de6f26ae:double,Composure_double_VectorAs...
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$11$$anon$1.hasNext(WholeStageCodegenExec.scala:619)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:220)
at org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:298)
at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1165)
at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1156)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1091)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1156)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:882)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:335)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:286)
at org.apache.spark.rdd.ZippedPartitionsRDD2.compute(ZippedPartitionsRDD.scala:89)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:121)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
Caused by: org.apache.spark.SparkException: Encountered null while assembling a row with handleInvalid = "keep". Consider
removing nulls from dataset or using handleInvalid = "keep" or "skip".
at org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler.scala:287)
at org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler.scala:255)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35)
at org.apache.spark.ml.feature.VectorAssembler$.assemble(VectorAssembler.scala:255)
at org.apache.spark.ml.feature.VectorAssembler$$anonfun$4.apply(VectorAssembler.scala:144)
at org.apache.spark.ml.feature.VectorAssembler$$anonfun$4.apply(VectorAssembler.scala:143)
... 29 more
Can someone please help me figure out the error.
From your log:
Caused by: org.apache.spark.SparkException: Encountered null while assembling a row with handleInvalid = "keep". Consider
removing nulls from dataset or using handleInvalid = "keep" or "skip".
You have null value in assembling row.
vecAssembler = VectorAssembler(inputCols=FEATURES_COL1, outputCol="features")
Please fill in null value in your data first, or use handleInvalid parameter of vector assembler, see Official Doc for detail.