Sparkで既存のRDDにRDDを追加するには？

Question

私はRDDを持っており、それにRDDを追加したいと思っています。スパークでどうすればいいですか？ 私は以下のようなコードを持っています。私が持っているdStreamからRDDを返したい。

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

私は、Apacheのスパークによって提供さJavaDStreamクラスの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が不変あるので、何を行うことができ、新しいRDDを作成するためにsparkContext.parallizeを使用して、新しいものを返すです。

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

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