我想在 Spark 中读取 CSV 并将其转换为 DataFrame 并将其存储在 HDFS 中，并带有df.registerTempTable("table_name")

我试过：

scala> val df = sqlContext.load("hdfs:///csv/file/dir/file.csv")

我得到的错误：

java.lang.RuntimeException: hdfs:///csv/file/dir/file.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)

在Apache Spark中将CSV文件加载为数据帧的正确命令是什么？

spark-csv 是核心 Spark 功能的一部分，不需要单独的库。所以你可以做例如

df = spark.read.format("csv").option("header", "true").load("csvfile.csv")

在scala中，(这适用于任何格式分隔符提及"，"对于csv，"\t"对于tsv等)

val df = sqlContext.read.format("com.databricks.spark.csv") .option("delimiter", ",") .load("csvfile.csv")

使用 Spark 2.x 解析 CSV 并加载为 DataFrame/DataSet

首先，初始化SparkSession对象，默认情况下它将在shell 中作为spark

val spark = org.apache.spark.sql.SparkSession.builder
.master("local") # Change it as per your cluster
.appName("Spark CSV Reader")
.getOrCreate;

使用以下任一方式将 CSV 加载为DataFrame/DataSet

1.以编程方式进行

val df = spark.read
.format("csv")
.option("header", "true") //first line in file has headers
.option("mode", "DROPMALFORMED")
.load("hdfs:///csv/file/dir/file.csv")

更新：从此处添加所有选项，以防将来链接断开

路径：文件的位置。类似于Spark可以接受标准的Hadoop通配表达式。
标头：设置为 true 时，文件的第一行将用于命名列，不会包含在数据中。所有类型都将被假定为字符串。默认值为 false。
：默认情况下列使用分隔符，但分隔符可以设置为任何字符
quote：默认情况下，引号字符为 "，但可以设置为任何字符。引号内的分隔符将被忽略
转义：默认情况下，转义字符为，但可以设置为任何字符。转义的引号字符将被忽略
parserLib：默认情况下，可以将">commons"设置为">univocity"以使用该库进行CSV解析。
：确定解析模式。默认情况下，它是宽松的。可能的值为：
- ALLOWIVE：尝试解析所有行：插入空值表示缺少标记，并忽略额外的标记。
- DROPMALFORM：删除标记少于或多于预期的行，或者删除与架构不匹配的标记
- FAILFAST：如果遇到任何格式错误的行，则中止并显示运行时异常字符集：默认为"UTF-8"，但可以设置为其他有效的字符集名称
架构：自动推断列类型。它需要对数据进行一次额外的传递，默认情况下为假
注释：跳过以此字符开头的行。默认值为"#"。通过将其设置为 null 来禁用注释。
nullValue：指定指示 null 值的字符串，任何与此字符串匹配的字段都将在数据帧中设置为 null
dateFormat：指定一个字符串，该字符串指示读取日期或时间戳时要使用的日期格式。自定义日期格式遵循 java.text.SimpleDateFormat 中的格式。这适用于日期类型和时间戳类型。默认情况下，它是空的，这意味着尝试通过java.sql.Timestamp.valueOf()和java.sql.Date.valueOf()解析时间和日期。

阿拉伯数字。您也可以以这种SQL方式执行此操作

val df = spark.sql("SELECT * FROM csv.`hdfs:///csv/file/dir/file.csv`")

依赖关系：

"org.apache.spark" % "spark-core_2.11" % 2.0.0,
"org.apache.spark" % "spark-sql_2.11" % 2.0.0,

Spark 版本 <2.0

val df = sqlContext.read
.format("com.databricks.spark.csv")
.option("header", "true") 
.option("mode", "DROPMALFORMED")
.load("csv/file/path");

依赖：

"org.apache.spark" % "spark-sql_2.10" % 1.6.0,
"com.databricks" % "spark-csv_2.10" % 1.6.0,
"com.univocity" % "univocity-parsers" % LATEST,

它的Hadoop是2.6，Spark是1.6，没有"databricks"包。

import org.apache.spark.sql.types.{StructType,StructField,StringType,IntegerType};
import org.apache.spark.sql.Row;
val csv = sc.textFile("/path/to/file.csv")
val rows = csv.map(line => line.split(",").map(_.trim))
val header = rows.first
val data = rows.filter(_(0) != header(0))
val rdd = data.map(row => Row(row(0),row(1).toInt))
val schema = new StructType()
.add(StructField("id", StringType, true))
.add(StructField("val", IntegerType, true))
val df = sqlContext.createDataFrame(rdd, schema)

使用 Spark 2.0，以下是阅读 CSV 的方法

val conf = new SparkConf().setMaster("local[2]").setAppName("my app")
val sc = new SparkContext(conf)
val sparkSession = SparkSession.builder
.config(conf = conf)
.appName("spark session example")
.getOrCreate()
val path = "/Users/xxx/Downloads/usermsg.csv"
val base_df = sparkSession.read.option("header","true").
csv(path)

在 Java 1.8 中此代码片段非常适合读取 CSV 文件

聚甲醛.xml

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.0.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql_2.10 -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.10</artifactId>
<version>2.0.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.scala-lang/scala-library -->
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.11.8</version>
</dependency>
<dependency>
<groupId>com.databricks</groupId>
<artifactId>spark-csv_2.10</artifactId>
<version>1.4.0</version>
</dependency>

爪哇岛

SparkConf conf = new SparkConf().setAppName("JavaWordCount").setMaster("local");
// create Spark Context
SparkContext context = new SparkContext(conf);
// create spark Session
SparkSession sparkSession = new SparkSession(context);
Dataset<Row> df = sparkSession.read().format("com.databricks.spark.csv").option("header", true).option("inferSchema", true).load("hdfs://localhost:9000/usr/local/hadoop_data/loan_100.csv");
//("hdfs://localhost:9000/usr/local/hadoop_data/loan_100.csv");
System.out.println("========== Print Schema ============");
df.printSchema();
System.out.println("========== Print Data ==============");
df.show();
System.out.println("========== Print title ==============");
df.select("title").show();

Penny 的 Spark 2 示例是在 spark2 中执行此操作的方法。还有一个技巧：通过对数据进行初始扫描，通过将选项inferSchema设置为true，为您生成该标头

然后，在这里，假设spark是您设置的火花会话，则是在亚马逊在 S3 上托管的所有 Landsat 图像的 CSV 索引文件中加载的操作。

/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements.  See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License.  You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
val csvdata = spark.read.options(Map(
"header" -> "true",
"ignoreLeadingWhiteSpace" -> "true",
"ignoreTrailingWhiteSpace" -> "true",
"timestampFormat" -> "yyyy-MM-dd HH:mm:ss.SSSZZZ",
"inferSchema" -> "true",
"mode" -> "FAILFAST"))
.csv("s3a://landsat-pds/scene_list.gz")

坏消息是：这会触发对文件的扫描;对于像这个 20+MB 压缩 CSV 文件这样大的东西，在长途连接中可能需要 30 秒。请记住这一点：一旦你进入架构，你最好手动编码架构。

(代码片段 Apache 软件许可证 2.0 许可以避免所有歧义;我作为 S3 集成的演示/集成测试所做的)

解析 CSV 文件有很多挑战，如果文件大小更大，如果列值中有非英语/转义/分隔符/其他字符，它就会不断累积，这可能会导致解析错误。

然后，魔力就在于使用的选项。那些对我有用并希望应该涵盖大多数边缘情况的代码在下面的代码中：

### Create a Spark Session
spark = SparkSession.builder.master("local").appName("Classify Urls").getOrCreate()
### Note the options that are used. You may have to tweak these in case of error
html_df = spark.read.csv(html_csv_file_path, 
header=True, 
multiLine=True, 
ignoreLeadingWhiteSpace=True, 
ignoreTrailingWhiteSpace=True, 
encoding="UTF-8",
sep=',',
quote='"', 
escape='"',
maxColumns=2,
inferSchema=True)

希望有帮助。有关更多参考：使用 PySpark 2 读取具有 HTML 源代码的 CSV

注意：上面的代码来自Spark 2 API，其中CSV文件读取API与可安装的Spark内置包捆绑在一起。

注意：PySpark 是 Spark 的 Python 包装器，与 Scala/Java 共享相同的 API。

如果你正在构建一个带有scala 2.11和Apache 2.0或更高版本的jar。

无需创建sqlContext或sparkContext对象。只需一个SparkSession对象就足以满足所有需求。

以下是工作正常的mycode：

import org.apache.spark.sql.{DataFrame, Row, SQLContext, SparkSession}
import org.apache.log4j.{Level, LogManager, Logger}
object driver {
def main(args: Array[String]) {
val log = LogManager.getRootLogger
log.info("**********JAR EXECUTION STARTED**********")
val spark = SparkSession.builder().master("local").appName("ValidationFrameWork").getOrCreate()
val df = spark.read.format("csv")
.option("header", "true")
.option("delimiter","|")
.option("inferSchema","true")
.load("d:/small_projects/spark/test.pos")
df.show()
}
}

如果您在集群中运行，只需在定义sparkBuilder对象时将.master("local")更改为.master("yarn")

Spark Doc涵盖了以下内容： https://spark.apache.org/docs/2.2.0/sql-programming-guide.html

使用 Spark 2.4+，如果你想从本地目录加载一个 csv，那么你可以使用 2 个会话并将其加载到 Hive 中。第一个会话应该使用 master() config 作为 "local[*]" 创建，第二个会话应该使用 "yarn" 和 Hive 启用。

下面的一个对我有用。

import org.apache.log4j.{Level, Logger}
import org.apache.spark._
import org.apache.spark.rdd._
import org.apache.spark.sql._
object testCSV { 
def main(args: Array[String]) {
Logger.getLogger("org").setLevel(Level.ERROR)
val spark_local = SparkSession.builder().appName("CSV local files reader").master("local[*]").getOrCreate()
import spark_local.implicits._
spark_local.sql("SET").show(100,false)
val local_path="/tmp/data/spend_diversity.csv"  // Local file
val df_local = spark_local.read.format("csv").option("inferSchema","true").load("file://"+local_path) // "file://" is mandatory
df_local.show(false)
val spark = SparkSession.builder().appName("CSV HDFS").config("spark.sql.warehouse.dir", "/apps/hive/warehouse").enableHiveSupport().getOrCreate()
import spark.implicits._
spark.sql("SET").show(100,false)
val df = df_local
df.createOrReplaceTempView("lcsv")
spark.sql(" drop table if exists work.local_csv ")
spark.sql(" create table work.local_csv as select * from lcsv ")
}

当与spark2-submit --master "yarn" --conf spark.ui.enabled=false testCSV.jar一起运行时，它运行良好并在 hive 中创建表。

将以下 Spark 依赖项添加到 POM 文件：

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.2.0</version>
</dependency>

火花配置：

val spark = SparkSession.builder().master("local").appName("Sample App").getOrCreate()

读取 csv 文件：

val df = spark.read.option("header", "true").csv("FILE_PATH")

显示输出：

df.show()

如果使用 Spark 2.0+ 试试这个

For non-hdfs file:
df = spark.read.csv("file:///csvfile.csv")

For hdfs file:
df = spark.read.csv("hdfs:///csvfile.csv")
For hdfs file (with different delimiter than comma:
df = spark.read.option("delimiter","|")csv("hdfs:///csvfile.csv")

注意：- 这适用于任何分隔文件。只需使用选项("分隔符"，)即可更改值。

希望这是有帮助的。

要从系统上的相对路径读取，请使用 System.getProperty 方法获取当前目录，并进一步使用相对路径加载文件。

scala> val path = System.getProperty("user.dir").concat("/../2015-summary.csv")
scala> val csvDf = spark.read.option("inferSchema","true").option("header", "true").csv(path)
scala> csvDf.take(3)

火花：2.4.4 斯卡拉：2.11.12

默认文件格式是带有spark.read..和文件读取csv的Parquet，这就是为什么你会得到异常。使用您尝试使用的 API 指定 csv 格式

使用内置的Spark csv，您可以使用Spark> 2.0的新SparkSession对象轻松完成它。

val df = spark.
read.
option("inferSchema", "false").
option("header","true").
option("mode","DROPMALFORMED").
option("delimiter", ";").
schema(dataSchema).
csv("/csv/file/dir/file.csv")
df.show()
df.printSchema()

您可以设置各种选项。

header：您的文件顶部是否包含标题行
inferSchema：是否要自动推断模式。默认值为true。我总是更喜欢提供架构来确保正确的数据类型。
mode：解析模式、允许模式、格式删除模式或快速故障
delimiter：要指定分隔符，默认值为逗号('，')

Spark - 将 CSV 文件加载为数据帧