I have dataset with one column (let say: empId) which can have large number of rows(18k-20k or more) and I am trying to use
Dataset<Row> allEmpIds=inputData;
allEmpIds.repartition(100).foreachPartition(rootPartition -> {
if (!rootPartition.hasNext()) {
return;
}
rootPartition.forEachRemaining(row -> {
....code to generate report for each empIds
});
But this is throwing error
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:416)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:406)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:163)
at org.apache.spark.SparkContext.clean(SparkContext.scala:2332)
at org.apache.spark.rdd.RDD.$anonfun$foreachPartition$1(RDD.scala:979)
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:385)
at org.apache.spark.rdd.RDD.foreachPartition(RDD.scala:978)
at org.apache.spark.sql.Dataset.$anonfun$foreachPartition$1(Dataset.scala:2741)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at org.apache.spark.sql.Dataset.$anonfun$withNewRDDExecutionId$1(Dataset.scala:3354)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:80)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:127)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:75)
at org.apache.spark.sql.Dataset.withNewRDDExecutionId(Dataset.scala:3350)
at org.apache.spark.sql.Dataset.foreachPartition(Dataset.scala:2741)
at org.apache.spark.sql.Dataset.foreachPartition(Dataset.scala:2752)
I am using repartition to run spark job in parallel for faster computation as for each empId I have to some computations which is time consuming. Can someone help here?
Thanks