我可以从Spark程序而不是从RDD编写纯文本HDFS（或本地）文件吗？

Question

我有一个Spark程序（在Scala中）和一个sparkcontext。我正在用rdd的saveastextfile编写一些文件。在我的本地机器上，我可以使用本地文件路径，它与本地文件系统一起工作。在我的集群上，它与HDFS一起工作。

我还想写其他任意文件作为处理的结果。我在本地计算机上将它们作为常规文件编写，但希望它们进入集群上的HDFS。

sparkcontext似乎有一些与文件相关的方法，但它们似乎都是输入而不是输出。

我该怎么做？

Answer 1

以下是对我最有效的方法（使用Spark2.0）：

val path = new Path("hdfs://namenode:8020/some/folder/myfile.txt")
val conf = new Configuration(spark.sparkContext.hadoopConfiguration)
conf.setInt("dfs.blocksize", 16 * 1024 * 1024) // 16MB HDFS Block Size
val fs = path.getFileSystem(conf)
if (fs.exists(path))
    fs.delete(path, true)
val out = new BufferedOutputStream(fs.create(path)))
val txt = "Some text to output"
out.write(txt.getBytes("UTF-8"))
out.flush()
out.close()
fs.close()

Answer 2

多亏了marios和kostya，但是从Spark将文本文件写入HDFS的步骤很少。

// Hadoop Config is accessible from SparkContext
val fs = FileSystem.get(sparkContext.hadoopConfiguration); 

// Output file can be created from file system.
val output = fs.create(new Path(filename));

// But BufferedOutputStream must be used to output an actual text file.
val os = BufferedOutputStream(output)

os.write("Hello World".getBytes("UTF-8"))

os.close()

注意，已经建议的fsDataOutputStream是Java序列化的对象输出流，而不是文本输出流。writeutf方法看起来是写plaint文本，但它实际上是一种包含额外字节的二进制序列化格式。