Convert iterable to RDD(将可迭代转换为RDD)
问题描述
我想从火花流到几个弹性搜索索引。
我创建了成对的<key(index), value>
,当我执行groupByKey时,结果是<key(index), Iterable<value>>
的元组,但为了使用ElasticSearch-Spark插件保存到ElasticSearch,我需要JavaRDD<value>
的值。
我知道有一个可以从List创建Java RDD的SparkConext.parallize(List)选项,但该选项只能在驱动程序上执行。
是否有其他选项可以创建可以在Executor上执行的JavaRDD?或者我可以实现Tuple2<key(index), JavaRDD<value>>
在Executor上工作的另一种方式?
如果没有,我如何才能在驱动程序上仅将Iterator切换到JavaRDD,并在Executor处将插件写入ElasticSearch?
谢谢,
Daniela
推荐答案
我想说的是,必须有如下所示的smth
JavaPairRDD<Key, Iterable<Value>> pair = ...;
JavaRDD<Iterable<Value>> values = pair.map(t2 -> t2._2());
JavaRDD<Value> onlyValues = values.flatMap(it -> it);
替代方法是
JavaPairRDD<Key, Iterable<Value>> pair = ...;
JavaRDD<Key, Value> keyValues = pair.flatMapValues(v1 -> v1);
JavaRDD<Value> values = keyValues.map(t2 -> t2._2());
这篇关于将可迭代转换为RDD的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持编程学习网!
本文标题为:将可迭代转换为RDD
基础教程推荐
- 在螺旋中写一个字符串 2022-01-01
- 如何强制对超级方法进行多态调用? 2022-01-01
- Java 中保存最后 N 个元素的大小受限队列 2022-01-01
- 由于对所需库 rt.jar 的限制,对类的访问限制? 2022-01-01
- 如何在不安装整个 WTP 包的情况下将 Tomcat 8 添加到 Eclipse Kepler 2022-01-01
- Spring Boot Freemarker从2.2.0升级失败 2022-01-01
- 首次使用 Hadoop,MapReduce Job 不运行 Reduce Phase 2022-01-01
- 如何使用 Stream 在集合中拆分奇数和偶数以及两者的总和 2022-01-01
- 如何对 HashSet 进行排序? 2022-01-01
- 如何使用 Eclipse 检查调试符号状态? 2022-01-01