在顶部/使用 Spark 保存和加载 JSON 和 scala 的对象

Question

我在使用 spark 读取和写入文件到“远程”文件系统（例如 hadoop）时遇到问题。

内容

1. 我在本地做了什么？
2. 我想在“远程”上做什么？

1。我在本地做了什么？

就目前而言，我在本地使用 spark - 向我的设备读取和写入文件，如下所示：

Spark-Session 初始化：

  val spark: SparkSession = Try(
    SparkSession.builder()
      .master("local[*]")
      .appName("app")
      .getOrCreate()) match {
    case Success(session)=>session
    case Failure(exception)=> throw new Exception(s"Failed initializing spark, due to: ${exception.getMessage}")
  }

在本地保存/写入，然后加载/读取：

（Json 文件）

  val content = "{"a": 10, "b": [], "c": {"x": "1", "z": {}}, {"x": "2", "z": {}}}"  // dummy JSON as string
  val fileName = "full_path/sample.json"

  // ... verify directory exists and create it if not ...

  // write sample.json with the content above:
  new PrintWriter(fileName) {
    write(content)
    close()
  }

  // Read & Operate on it:
  val jsonAsBufferedSource = Source.fromFile(fileName)

（Scala 的案例类）

  case class Dummy(string: String, i: Int) extends Serializable {}
  val content = Dummy("42 is the best number", 42)       // Dummy instance
  val fileName = "full_path/sample.dummy"               // 'dummy' is the serialized saved-object name.
  
  // ... verify directory exists and create it if not ...

  // Write it:
  val output = new ObjectOutputStream(new FileOutputStream(fileName))
  output.writeObject(content)
  output.close()

  // Read:
  val input = new ObjectInputStream(new FileInputStream(fileName))
  val dummyObject = input.readObject.asInstanceOf[Dummy]
  input.close()

  // Operate:
  dummyObject.i   // 42

2。我想在“远程”上做什么？

我希望能够使用 spark 在 HDFS、S3 或任何其他可用的“远程”文件系统上读取/写入，就像我在本地所做的那样。

我的问题主要是：

• Spark 配置：应该更改什么以及如何更改？ [大师等..]
• 使用 Spark：
  • 如何像在本地一样保存和加载可序列化对象？
  • 如何保存 Json 字符串，并将其作为 BufferedSource 加载？

一般来说 - 我想让自己在本地/远程使用我的应用程序的相同“内部接口”工作。

感谢您的阅读！

编辑

我希望我的应用程序在测试和调试时将文件保存/读取到磁盘并在我的计算机磁盘上工作。我希望它在生产时使用远程文件系统保存/读取。
是否可以使用相同的火花方法？使用什么火花配置？

欧伦

Answer 1

不确定我是否理解这个问题。 Spark 与 file:// 和 hdfs:// 或 s3a:// 前缀相同。错误的是 Source.fromFile 和 PrintWriter

您需要重写函数以使用正确的 Spark 方法，因为 Spark 旨在在集群中运行，而不是与一台机器隔离（称为驱动程序）

// read all JSON files in a folder
val df = spark.read.json("file:///path/to/full_path/")

// write the dataframe to HDFS folder
df.write.format("json").save("hdfs://namenode.fqdn:port/hdfs/path/")

当然，您可以序列化一个类，“本地”写入文件（deploy-mode=cluster 时将是“远程”），然后上传那个，但这看起来不像你在这里做。而不是这样做，您将parellelize 序列化对象的Seq。

使用 json4s 而不是 ObjectOutputStream 从案例类中获取 JSON。