如何向 Spark 中现有的 RDD 添加更多的 RDD？

Question

我有一个 RDD，想向其中添加更多的 RDD。我怎样才能在 Spark 中做到这一点？我有如下代码。我想从我拥有的 dStream return RDD。

JavaDStream<Object> newDStream = dStream.map(this);
JavaRDD<Object> rdd = context.sparkContext().emptyRDD();
return newDStream.wrapRDD(context.sparkContext().emptyRDD());

我没有找到太多关于 Apache Spark 提供的 JavaDStream class 方法的 wrapRDD 文档。

Answer 1

您可以使用 JavaStreamingContext.queueStream 并用 Queue<RDD<YourType>>:

填充

public JavaInputDStream<Object> FillDStream() {
    LinkedList<RDD<Object>> rdds = new LinkedList<RDD<Object>>();
    rdds.add(context.sparkContext.emptyRDD());
    rdds.add(context.sparkContext.emptyRDD());

    JavaInputDStream<Object> filledDStream = context.queueStream(rdds);
    return filledStream;
}

Answer 2

由于 RDD 是 不可变的，您可以做的是使用 sparkContext.parallelize 创建一个新的 RDD 并 return 新的。

List<Object> objectList = new ArrayList<Object>;
objectList.add("your content");

JavaRDD<Object> objectRDD = sparkContext.parallelize(objectList);
JavaRDD<Object> newRDD = oldRDD.union(objectRDD);

见https://spark.apache.org/docs/latest/rdd-programming-guide.html#parallelized-collections

如何向 Spark 中现有的 RDD 添加更多的 RDD？

How to add more RDD to existing RDD in Spark?

java

apache-spark

spark-streaming