scala – Spark – 如何读取csv文件作为DataFrame？

发布时间：2020-12-16 09:46:50 所属栏目：安全来源：网络整理

导读：我想在spark中读取CSV，我想将其存储在spark数据框中。 https://databricks.com/blog/2015/02/17/introducing-dataframes-in-spark-for-large-scale-data-science.html 我努力了： scala val df = sqlContext.load("/home/donbeo/Documents/dataset/winequal

我想在spark中读取CSV，我想将其存储在spark数据框中。 https://databricks.com/blog/2015/02/17/introducing-dataframes-in-spark-for-large-scale-data-science.html

我努力了：

scala> val df = sqlContext.load("/home/donbeo/Documents/dataset/winequality-red.csv")
java.lang.RuntimeException: file:/home/donbeo/Documents/dataset/winequality-red.csv is not a Parquet file. expected magic number at tail [80,65,82,49] but found [49,59,54,10]
    at parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:418)
    at org.apache.spark.sql.parquet.ParquetRelation2$MetadataCache$$anonfun$refresh$6.apply(newParquet.scala:277)
    at org.apache.spark.sql.parquet.ParquetRelation2$MetadataCache$$anonfun$refresh$6.apply(newParquet.scala:276)
    at scala.collection.parallel.mutable.ParArray$Map.leaf(ParArray.scala:658)
    at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply$mcV$sp(Tasks.scala:54)
    at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:53)
    at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:53)
    at scala.collection.parallel.Task$class.tryLeaf(Tasks.scala:56)
    at scala.collection.parallel.mutable.ParArray$Map.tryLeaf(ParArray.scala:650)
    at scala.collection.parallel.AdaptiveWorkStealingTasks$WrappedTask$class.compute(Tasks.scala:165)
    at scala.collection.parallel.AdaptiveWorkStealingForkJoinTasks$WrappedTask.compute(Tasks.scala:514)
    at scala.concurrent.forkjoin.RecursiveAction.exec(RecursiveAction.java:160)
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

但它不工作。
什么是正确的命令？

解决方法

Spark SQL仅为3种类型的数据源提供内置支持：

> Parquet(这是默认值)
> Json
> Jdbc

对于CSV，有一个单独的库：spark-csv

它的CsvContext类提供了可以用于加载csv的csvFile方法。

val cars = sqlContext.csvFile("cars.csv") // uses implicit class CsvContext

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!