在 spark 版本 1.5.1 中保存为文本文件时，我使用：rdd.saveAsTextFile('<drectory>')。
但是如果我想在那个目录中找到文件，我该如何命名呢？
目前，我认为它被命名为part-00000，它必须是一些默认值。我该如何给它起名字？
【问题讨论】：
标签： apache-spark pyspark rdd
这个问题的正确答案是saveAsTextFile 不允许你命名实际的文件。
这样做的原因是数据是分区的，并且在作为参数提供给saveAsTextFile(...) 调用的路径内，它会将其视为一个目录，然后每个分区写入一个文件。
你可以调用rdd.coalesce(1).saveAsTextFile('/some/path/somewhere')，它会创建/some/path/somewhere/part-0000.txt。
如果您需要比这更多的控制，则需要在执行rdd.collect() 后执行实际的文件操作。
注意，这会将所有数据提取到一个执行程序中，因此您可能会遇到内存问题。这就是你要承担的风险。
【讨论】：
正如我在上面的评论中所说，可以在here 找到带有示例的文档。并引用saveAsTextFile方法的描述：
将此RDD保存为文本文件，使用元素的字符串表示。
在以下示例中，我将一个简单的 RDD 保存到文件中，然后加载并打印其内容。
samples = sc.parallelize([
("abonsanto@fakemail.com", "Alberto", "Bonsanto"),
("mbonsanto@fakemail.com", "Miguel", "Bonsanto"),
("stranger@fakemail.com", "Stranger", "Weirdo"),
("dbonsanto@fakemail.com", "Dakota", "Bonsanto")
])
print samples.collect()
samples.saveAsTextFile("folder/here.txt")
read_rdd = sc.textFile("folder/here.txt")
read_rdd.collect()
输出将是
('abonsanto@fakemail.com', 'Alberto', 'Bonsanto')
('mbonsanto@fakemail.com', 'Miguel', 'Bonsanto')
('stranger@fakemail.com', 'Stranger', 'Weirdo')
('dbonsanto@fakemail.com', 'Dakota', 'Bonsanto')
[u"('abonsanto@fakemail.com', 'Alberto', 'Bonsanto')",
u"('mbonsanto@fakemail.com', 'Miguel', 'Bonsanto')",
u"('stranger@fakemail.com', 'Stranger', 'Weirdo')",
u"('dbonsanto@fakemail.com', 'Dakota', 'Bonsanto')"]
让我们看看使用基于 Unix 的终端。
usr@host:~/folder/here.txt$ cat *
('abonsanto@fakemail.com', 'Alberto', 'Bonsanto')
('mbonsanto@fakemail.com', 'Miguel', 'Bonsanto')
('stranger@fakemail.com', 'Stranger', 'Weirdo')
('dbonsanto@fakemail.com', 'Dakota', 'Bonsanto')
【讨论】：
不可能像@nod 所说的那样命名文件。但是，之后可以立即重命名文件。使用 PySpark 的示例：
sc._jsc.hadoopConfiguration().set(
"mapred.output.committer.class",
"org.apache.hadoop.mapred.FileOutputCommitter")
URI = sc._gateway.jvm.java.net.URI
Path = sc._gateway.jvm.org.apache.hadoop.fs.Path
FileSystem = sc._gateway.jvm.org.apache.hadoop.fs.FileSystem
fs = FileSystem.get(URI("s3://{bucket_name}"), sc._jsc.hadoopConfiguration())
file_path = "s3://{bucket_name}/processed/source={source_name}/year={partition_year}/week={partition_week}/"
# remove data already stored if necessary
fs.delete(Path(file_path))
df.saveAsTextFile(file_path, compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")
# rename created file
created_file_path = fs.globStatus(Path(file_path + "part*.gz"))[0].getPath()
fs.rename(
created_file_path,
Path(file_path + "{desired_name}.jl.gz"))
【讨论】：