将 RDD 保存到 Scala 中的文本文件

Posted 2023-04-15

【中文标题】将 RDD 保存到 Scala 中的文本文件

【英文标题】：Saving a RDD to a text file in Scala

【发布时间】：2018-01-20 07:22:29

【问题描述】：

我想做两件事：

将RDD splitRDD 的内容显示到控制台。 将结果保存到文本文件中。

下面的第 3 行 scala 代码打印出密钥，但我正在寻找值。

val emailMsg = sc.textFile(file);`
val splitRDD = emailMsg.map( line => line.split("."));
splitRDD.foreach(println);
splitRDD.coalesce(1).saveAsTextFile("newfile")

【问题讨论】：

能否请您添加 splitRDD 的架构。

不确定“下面的第 3 行 scala 代码打印出密钥”是什么意思。 split 方法输出一个数组，其toString 方法实际上不会打印数组本身的任何成员。如果您只想打印第二项（例如），您应该执行以下操作：splitRDD.foreach(row => row(1))。

你为什么要line => line.split(".")？你能给出示例文件输入和你预期的打印输出吗？

感谢您的帮助。我正在对电子邮件消息（文本文件）进行拆分，并尝试将发件人、收件人、日期和主题分开。所以，我正在寻找诸如“主题：”和“收件人：”之类的词。

【参考方案1】：

我会假设你的文件看起来像这样

key1.value1
key2.value2

并且您想以其他格式打印和保存 values 或 pairs。

如果您只想打印和保存值，您可以将 splitRDD 转换为仅值 RDD。

val valRDD = splitRDD.map( _( 1 ) )
valRDD.foreach( println )

请注意，saveAsTextFile 不会以易于使用的格式保存文件，因此您可能需要一个简单的文本编写器（Java PrintWriter 就可以了）。

以两种不同格式打印和保存splitRDD 的示例

import org.apache.spark._
import java.io. PrintWriter, File, FileOutputStream 

...

val pwText = new PrintWriter(
    new File( "emailMsgValues.txt" )
)

val pwCSV = new PrintWriter(
    new File( "emailMsgPair.csv" )
)

val emailMsg = sc.textFile( "data/emailMsg.txt" )

val splitRDD = emailMsg.map( line => line.split( '.' ) )

println( "Printing and writing values in text" )

val valRDD = splitRDD.map( _( 1 ) ).collect()

valRDD.foreach( value => 

    println( value )
    pwText.write( value + "\n" )
 )

println( "Printing and writing pairs in csv" )

splitRDD.collect().foreach( pair => 

    println( pair.mkString( "," ) )
    pwCSV.write( pair.mkString( "," ) + "\n" )

 )

pwText.close()
pwCSV.close()

【讨论】：

非常好，正是我需要的。

太棒了！如果有帮助，您可以接受并投票赞成答案。干杯@SteveMcAffer

没问题，只要关注这个accept guide 和upvote guide。 @SteveMcAffer

【参考方案2】：

你说的第三行是打印不是关键。它实际上是打印数组对象，像这样

[Ljava.lang.String;@384efaf
[Ljava.lang.String;@5bc8b97c
[Ljava.lang.String;@18194125
[Ljava.lang.String;@364838ab
[Ljava.lang.String;@254b1df2

您需要做的是将该 Array 对象转换为 spark sql 的 Row 对象。所以用

import org.apache.spark.sql.Row

并像这样更改代码的第二行

val splitRDD = emailMsg.map( line => Row.fromSeq(line.split(" ")))

【讨论】：

感谢您的帮助。这太棒了！