java - Spark-Shell:org.apache.spark.SparkException:任务无法序列化

Question

我正在尝试使用Oracle虚拟机通过Spark将HDFS [/tmp/chicago_test_load/chicago_crimes_01_present.csv]中的Chicago Crime数据附加到Hortonworks Sandbox的配置单元表中。

Last login: Mon Feb 18 06:27:47 2019 from 172.18.0.3
[maria_dev@sandbox-hdp ~]$ spark-shell
SPARK_MAJOR_VERSION is set to 2, using Spark2
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://sandbox-hdp.hortonworks.com:4040
Spark context available as 'sc' (master = local[*], app id = local-1550484326924).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.3.0.2.6.5.0-292
      /_/

Using Scala version 2.11.8 (OpenJDK 64-Bit Server VM, Java 1.8.0_171)
Type in expressions to have them evaluated.
Type :help for more information.

import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._

val spark = SparkSession.builder().appName("StatsAnalyzer").enableHiveSupport().config("hive.exec.dynamic.partition","true").config("hive.exec.dynamic.partition.mode", "nonstrict").getOrCreate()
val sqlContext = new HiveContext(sc)
var dr = sqlContext.read.format("com.databricks.spark.csv").option("delimeter", ",")

var df = dr.load("/tmp/chicago_test_load/chicago_crimes_01_present.csv")

var header = df.first()

df = df.filter(row => row != header)

当我执行 df.show()来详细查看数据框时，下面出现错误。

scala> df.show()
org.apache.spark.SparkException: Task not serializable
  at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:345)
  at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:335)
  at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:159)
  at org.apache.spark.SparkContext.clean(SparkContext.scala:2299)
  at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1.apply(RDD.scala:844)
  at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1.apply(RDD.scala:843)
  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.mapPartitionsWithIndex(RDD.scala:843)
  at org.apache.spark.sql.execution.WholeStageCodegenExec.doExecute(WholeStageCodegenExec.scala:608)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.SparkPlan.getByteArrayRdd(SparkPlan.scala:247)
  at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:337)
  at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:38)
  at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collectFromPlan(Dataset.scala:3272)
  at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2484)
  at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2484)
  at org.apache.spark.sql.Dataset$$anonfun$52.apply(Dataset.scala:3253)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
  at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3252)
  at org.apache.spark.sql.Dataset.head(Dataset.scala:2484)
  at org.apache.spark.sql.Dataset.take(Dataset.scala:2698)
  at org.apache.spark.sql.Dataset.showString(Dataset.scala:254)
  at org.apache.spark.sql.Dataset.show(Dataset.scala:723)
  at org.apache.spark.sql.Dataset.show(Dataset.scala:682)
  at org.apache.spark.sql.Dataset.show(Dataset.scala:691)
  ... 51 elided
Caused by: java.io.NotSerializableException: org.apache.spark.sql.DataFrameReader
Serialization stack:
        - object not serializable (class: org.apache.spark.sql.DataFrameReader, value: org.apache.spark.sql.DataFrameReader@8dedec8)
        - field (class: $iw, name: dr, type: class org.apache.spark.sql.DataFrameReader)
        - object (class $iw, $iw@485c8d5e)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@8dc73c7)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@363502a4)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@6112390a)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@2f541378)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@7d6e3e42)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@5fde8cf4)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@44556fbd)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@4d36c8cb)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@f03dc9b)
        - field (class: $line19.$read, name: $iw, type: class $iw)
        - object (class $line19.$read, $line19.$read@772ccecf)
        - field (class: $iw, name: $line19$read, type: class $line19.$read)
        - object (class $iw, $iw@41c02b8c)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@520dc647)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@6e52877d)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@4273c8be)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@62c63900)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@17a364)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@19b0f61c)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@b230f87)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@300788d6)
        - field (class: $line20.$read, name: $iw, type: class $iw)
        - object (class $line20.$read, $line20.$read@1eb9ab8f)
        - field (class: $iw, name: $line20$read, type: class $line20.$read)
        - object (class $iw, $iw@4ce1292a)
        - field (class: $iw, name: $outer, type: class $iw)
        - object (class $iw, $iw@1177049c)
        - field (class: $anonfun$1, name: $outer, type: class $iw)
        - object (class $anonfun$1, <function1>)
        - element of array (index: 2)
        - array (class [Ljava.lang.Object;, size 4)
        - field (class: org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10, name: references$1, type: class [Ljava.lang.Object;)
        - object (class org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10, <function2>)
  at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
  at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
  at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
  at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:342)
  ... 82 more

任何帮助表示赞赏。

Answer 1

不能这样做，将数据帧传递给Worker!通过 header 的。

有一些注意事项，如果路径中有多个文件怎么办？

尝试这种方法，我认为这要容易得多，您应该使用val not var:

val dfother = spark.read.option("header", "true").csv("file path").toDF("col1") 
dfother.show(false)

您也可以阅读为文本文件，并删除第一个记录，然后再拆分。在这里，您会看到2个文本文件，在这种情况下，前两行都已删除。

var ds = spark.read.text("/FileStore/tables/fff02.txt", "/FileStore/tables/fff01.txt").as[String].mapPartitions(_.drop(1))//.toDF

他们说，有多种方法可以给猫咪剥皮，但是请尝试使用标准方法。