scala-我可以从spark程序而不是从RDD写一个纯文本 HDFSS(或 local)文件吗?


0

我有一个Spark程序(在scala中)和一个SparkContext。我正在用RDD的saveastextfile编写一些文件。在 local计算机上,我可以使用 local文件路径,它与 local文件系统一起 job。在我的集群上,它与 HDFSSS一起 job。

我还想写其他任意文件作为处理的结果。我将它们作为常规文件写入 local计算机,但希望它们进入集群中的 HDFSSS。

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

我该怎么做?

4 答案

0

多亏了Marios和Kostya,但要从Spark将文本文件写入 HDFSSS还需要几个步骤。

// 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()

注意,这是一个Java序列化的对象输出流,而不是文本输出流。writeutf方法似乎写纯文本,但实际上它是一种包含额外字节的二进制序列化格式。


0

以下是最适合我的(使用Spark 2.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     HDFSS 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()

0

使用hdfs api(hadoop hdfs.jar),可以为hdfs路径创建inputstream/outputstream,并使用常规java.io类读取/写入文件。例如:

URI uri = URI.create (“hdfs://host:port/file path”);
Configuration conf = new Configuration();
FileSystem file = FileSystem.get(uri, conf);
FSDataInputStream in = file.open(new Path(uri));

此代码也可用于 local文件(将hdfs://更改为file://)。


0

将文件写入 HDFSSS的一个简单方法是使用序列文件。这里使用的是 localHadoop API,而不是Spark提供的API。

下面是一个简单的片段(在scala中):

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs._
import org.apache.hadoop.io._ 

val conf = new Configuration() // Hadoop configuration
val sfwriter = SequenceFile.createWriter(conf,
SequenceFile.Writer.file(new Path("hdfs://nn1.example.com/file1")),
SequenceFile.Writer.keyClass(LongWritable.class),
SequenceFile.Writer.valueClass(Text.class))
val lw = new LongWritable()
val txt = new Text()
lw.set(12)
text.set("hello")
sfwriter.append(lw, txt)
sfwriter.close()
...

如果没有密钥,则可以使用NullWritable.Class代替它:

SequenceFile.Writer.keyClass(NullWritable.class)
sfwriter.append(NullWritable.get(), new Text("12345"));

我来回答