我可以使用spack-csv将表示为字符串的CSV读取到Apache Spark中吗

Question

我知道如何使用spark csv将csv文件读入spark(https://github.com/databricks/spark-csv)，但我已经将csv文件表示为字符串，并希望将此字符串直接转换为dataframe。这可能吗？

Answer 1

公认的答案在spark 2.2.0中对我不起作用，但可以通过csvData找到我需要的东西。线toList

val fileUrl = getClass.getResource(s"/file_in_resources.csv")
val stream = fileUrl.getContent.asInstanceOf[InputStream]
val streamString = Source.fromInputStream(stream).mkString

val csvList = streamString.lines.toList

spark.read
  .option("header", "true")
  .option("inferSchema", "true")
  .csv(csvList.toDS())
  .as[SomeCaseClass]  

Answer 2

您可以使用，例如scala csv，将字符串解析为csv：

val myCSVdata:Array[列表[字符串]]=myCSVString。拆分（“\n”）。平面图（CSVParser.parseLine（_））

在这里，您可以进行更多的处理、数据清理、验证每一行是否解析良好以及是否具有相同数量的字段等。。。

然后，您可以将其作为记录的RDD：

val myCSVRDD： RDD[List[String]]=parkContext.parallelize（msCSVdata）

在这里，您可以将字符串列表转换为case类，以更好地反映csv数据的字段。在本例中，您应该从Person的创作中获得一些灵感：

https://spark.apache.org/docs/latest/sql-programming-guide.html#inferring-the-schema-using-reflection

我省略了这一步。

然后可以转换为数据帧：

导入spark。隐含。\umyCSVDataframe=myCSVRDD。toDF（）

Answer 3

更新：从Spark 2.2开始。最后，有一种正确的方法可以使用Dataset来完成。

import org.apache.spark.sql.{Dataset, SparkSession}
val spark = SparkSession.builder().appName("CsvExample").master("local").getOrCreate()

import spark.implicits._
val csvData: Dataset[String] = spark.sparkContext.parallelize(
  """
    |id, date, timedump
    |1, "2014/01/01 23:00:01",1499959917383
    |2, "2014/11/31 12:40:32",1198138008843
  """.stripMargin.lines.toList).toDS()

val frame = spark.read.option("header", true).option("inferSchema",true).csv(csvData)
frame.show()
frame.printSchema()

旧火花版本

实际上你可以，尽管它使用的是库的内部结构，并且没有被广泛宣传。只需创建和使用您自己的CsvParser实例。下面的示例适用于我的Spark 1.6.0和Spark-csv_2.10-1.4.0

    import com.databricks.spark.csv.CsvParser

val csvData = """
|userid,organizationid,userfirstname,usermiddlename,userlastname,usertitle
|1,1,user1,m1,l1,mr
|2,2,user2,m2,l2,mr
|3,3,user3,m3,l3,mr
|""".stripMargin
val rdd = sc.parallelize(csvData.lines.toList)
val csvParser = new CsvParser()
  .withUseHeader(true)
  .withInferSchema(true)


val csvDataFrame: DataFrame = csvParser.csvRdd(sqlContext, rdd)