将可迭代转换为 RDD

Question

我想从 spark-streaming 保存到几个弹性搜索索引。 我创建了一对<key(index), value>，当我执行groupByKey时，结果是<key(index), Iterable<value>>的元组，但是为了使用elasticsearch-spark插件保存到elasticsearch，我需要JavaRDD<value>的值。

我知道 sparkContext.parallelize(list) 有一个选项可以从列表中创建 JavaRDD，但这只能在驱动程序上执行。

是否有其他选项可以创建可以在执行器上执行的 JavaRDD？或者我可以实现Tuple2<key(index), JavaRDD<value>> 的另一种方式，它适用于执行程序？ 如果不是，我怎样才能只在驱动程序上将 Iterator 切换到 JavaRDD，并在执行程序上将插件写入 elasticsearch？

谢谢，

丹妮拉

Answer 1

我会说它必须有可能像下面这样

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