尝试将数据帧数据写入 CSV 文件。在火花

Question

每当我尝试运行我的代码时都会收到此错误。

(ERROR Executor: 0.0阶段任务0.0异常)

我的代码：

    import org.apache.log4j.{Level, Logger}
    import org.apache.spark._
    import org.apache.spark.sql.SparkSession
    
    
    object savingCSV extends App {
    
      // 2. defined Shema
      case class Person(ID: Int, NAME: String, SALARY: Int, CITY: String)
    
      Logger.getLogger("org").setLevel(Level.ERROR)
    
      val sc = new SparkContext("local[*]", "savingCSV")
      // 1. reading file as RDD
      val data = sc.textFile("data/readCSV.csv")
    
      // 3. Creating a DF
      val spark = SparkSession
        .builder
        .appName("savingCSV")
        .master("local[*]")
        .getOrCreate()
    
      import spark.implicits._
      // 4. saving DF as CSV in localDisk
      val RDDToDF = data.toDF()
    
      RDDToDF.write.format("csv").save("D:/Users/sarve/csvData/DFnewdata")
    
      spark.stop()
}

错误声明：

Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
21/08/12 16:44:48 ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0)
ExitCodeException exitCode=-1073741515:

Answer 1

Spark 提供了多层次的功能和相应的 API。

spark-core 带有SparkContext，它提供RDD 级别的功能。

spark-sql 添加了SparkSession，它提供了DataFrame、DataSet 和SQL 相关功能。

如果你只需要RDD api，那么你可以只使用SparkContext。

import org.apache.spark.{SparkConf, SparkContext}

object ReadAndExportCsv extends App {

  // create sparkConfig to define spark configuration
  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ReadAndExportCsv")

  // use config to create SparkContext
  val sparkContext = new SparkContext(sparkConf)

  //generally sparkContext is named as sc
  val sc = sparkContext

  // It will just reading the text file as a collection of lines.
  // It does not know about special properties of content of these lines.
  // So, it will be an RDD containing lines of the file simple RDD[String]
  val linesRdd = sc.textFile("/full_input_path/sample_1.csv")

  // lets append something to those lines
  val appendedLinesRdd = linesRdd.map(line => line + ", Yahoo! Learning basics of Spark")

  // lets write this modified RDD to a file
  appendedLinesRdd.saveAsTextFile("/full_output_path/sample_1")

  sparkContext.stop()
}

如果您需要 DataFrame，您将不得不使用 SparkSession，

import org.apache.spark.sql.SparkSession
import org.apache.spark.SparkConf

object ReadAndExportCsv2 extends App {

  // use camelCase for variable names in Scala. So id instead of ID, name instead of NAME
  final case class Person(id: Int, name: String, salary: Int, city: String)

  // create sparkConfig to define spark configuration
  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ReadAndExportCsv")

  // use config to create SparkSession
  val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
  //generally sparkSession is named as spark
  val spark = sparkSession

  // every sparkSession also contains a SparkContext
  val sparkContext = spark.sparkContext
  //generally sparkContext is named as sc
  val sc = sparkContext

  import spark.implicits._

  // spark session comes with more "intelligent" methods to read data.
  // It knows how to read CSV files.
  val csvDataFrame =
    spark.read.format("csv")
      .option("header", true) // tells spark to treat first line as header
      .option("inferSchema", true) // tells spark to try to infer types for each column by looking at values
      .load("/full_input_path/sample_1.csv")

  // The file contains header and we told spark to use header
  // So, Spark already knows that each record columns - ID,NAME,SALARY,CITY
  // ------------
  // Spark uses encoders to map these records to case classes.
  // Here since our case class fileds have simple one to one correspondence with columns
  // ID -> id, NAME -> name. SALARY -> salary, CITY -> city
  // So, Spark should be able to generate that encoder for Person
  // ------------
  // Hence we should be able convert this DataFrame into DataSet of Person objects
  val personDataSet = csvDataFrame.as[Person]

  personDataSet.write.format("csv")
    .option("header", true)
    .save("/full_output_path/sample_1")

  spark.close()
}