将RDD列表转换为spark中单个元素的RDD答案

【问题标题】：Convert RDD List to RDD of individual element in spark将RDD列表转换为spark中单个元素的RDD
【发布时间】：2017-06-09 00:08:21
【问题描述】：

我有一个输入 rdd (JavaRDD<List<String>>)，我想将其转换为 JavaRDD<String> 作为输出。
输入 RDD 列表的每个元素都应该成为输出 rdd 中的一个单独元素。

如何在java中实现？

JavaRDD<List<String>> input; //suppose rdd length is 2
input.saveAsTextFile(...)

输出：

[a,b]
[c,d]

我想要什么：

a
b
c
d

【问题讨论】：

不，flatmap 不起作用，因为它将再次包含元素列表，并且当我存储它时，它不会将每个元素存储在单独的行中。我的输入 RDD 同样是平面图的输出。
你不应该这么粗鲁，我是来帮忙的。顺便说一句，如果这是您的想法，我没有拒绝您。然而，这就是你想要的，但在可爱的Scala 中。 A published notebook
我是 spark 新手，如果我的评论听起来粗鲁，我深表歉意。
我想在rdd之上调用parallelize函数。该怎么做？
RDD 已经并行化了！！！

标签： java apache-spark

【解决方案1】：

将其转换为 DataFrame 并使用 Explode UDF 函数。

【讨论】：

【解决方案2】：

我使用下面的代码 sn-p 做了一个解决方法：

使用分隔符'\n' 连接列表的每个元素，然后使用标准 spark API 保存 rdd。

inputRdd.map(new Function<List<String>, String>() {
           @Override
           public String call(List<String> scores) throws Exception {
               int size = scores.size();
               StringBuffer sb = new StringBuffer();
               for (int i=0; i <size;i++){
                   sb.append(scores.get(i));
                   if(i!=size-1){
                        sb.append("\n");
                    }
               }
               return sb.toString();
           }
       }).saveAsTextFile("/tmp/data"));

【讨论】：

【解决方案3】：

如果 rdd 类型是RDD[List[String]]，你可以这样做：

val newrdd = rdd.flatmap(line => line)

每个元素都将成为新 rdd 中的新行。

【讨论】：

【解决方案4】：

以下将解决您的问题

 var conf = new SparkConf().setAppName("test")
            .setMaster("local[1]")
            .setExecutorEnv("executor-cores", "2")
 var sc = new SparkContext(conf)
 val a = sc.parallelize(Array(List("a", "b"), List("c", "d")))  
 a.flatMap(x => x).foreach(println)

输出：

一个
b

d

【讨论】：

我的数据集非常大，我想将 rdd 保存到文件中。
从 flatpMap 函数中我可以将每条记录写入文件吗？