在火花中解析 json

Question

我正在使用 json scala 库在 Spark 作业中解析本地驱动器中的 json：

val requestJson=JSON.parseFull(Source.fromFile("c:/data/request.json").mkString)
    val mainJson=requestJson.get.asInstanceOf[Map[String,Any]].get("Request").get.asInstanceOf[Map[String,Any]]
    val currency=mainJson.get("currency").get.asInstanceOf[String]

但是当我尝试通过指向 hdfs 文件位置来使用相同的解析器时，它不起作用：

val requestJson=JSON.parseFull(Source.fromFile("hdfs://url/user/request.json").mkString)

并给了我一个错误：

java.io.FileNotFoundException: hdfs:/localhost/user/request.json (No such file or directory)
  at java.io.FileInputStream.open0(Native Method)
  at java.io.FileInputStream.open(FileInputStream.java:195)
  at java.io.FileInputStream.<init>(FileInputStream.java:138)
  at scala.io.Source$.fromFile(Source.scala:91)
  at scala.io.Source$.fromFile(Source.scala:76)
  at scala.io.Source$.fromFile(Source.scala:54)
  ... 128 elided

如何使用 Json.parseFull 库从 hdfs 文件位置获取数据？

谢谢

Answer 1

Spark确实对JSON文档解析有内置的支持，这将在spark-sql_${scala.version} jar中提供。

在 Spark 2.0+ 中：

import org.apache.spark.sql.SparkSession 
val spark: SparkSession = SparkSession.builder.master("local").getOrCreate
val df = spark.read.format("json").json("json/file/location/in/hdfs")
df.show()

使用df对象，您可以对其执行所有支持的SQL操作，并且其数据处理将分布在节点之间，而requestJson 将仅在单机中计算。

Maven 依赖项

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

---

编辑：（根据注释从hdfs读取文件）

val hdfs = org.apache.hadoop.fs.FileSystem.get(
             new java.net.URI("hdfs://ITS-Hadoop10:9000/"), 
             new org.apache.hadoop.conf.Configuration()
           )
val path=new Path("/user/zhc/"+x+"/")
val t=hdfs.listStatus(path)
val in =hdfs.open(t(0).getPath)
val reader = new BufferedReader(new InputStreamReader(in))
var l=reader.readLine()

代码信用：来自另一个 SO 问题

Maven 依赖项：

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-hdfs</artifactId>
    <version>2.7.2</version> <!-- you can change this as per your hadoop version -->
</dependency>

Answer 2

在 Spark 2.0 中

更容易

val df = spark.read.json("json/file/location/in/hdfs")
df.show()

Answer 3

可以在Spark中使用以下内容从HDFS读取文件：val jsonText = sc.textFile（"hdfs://url/user/request.json"）.collect.mkString（""）