JavaDStreamをRDDに変換するには？または、JavaDStreamのマップ関数内に新しいRDDを作成する方法はありますか？

Question

kafkaから取得しているストリーミングデータは、hdfsファイルのパスであり、そのファイルのデータを取得する必要があります。

batchInputDStream.map(new Function<Tuple2<String,String>, FreshBatchInput>() { 

      @Override 
      public String call(Tuple2<String, String> arg0) 
        throws Exception { 
       StringReader reader = new StringReader(arg0._2); 
       JAXBContext jaxbContext = JAXBContext.newInstance(FreshBatchInput.class); 

        Unmarshaller jaxbUnmarshaller = jaxbContext.createUnmarshaller(); 
        FreshBatchInput input = (FreshBatchInput)jaxbUnmarshaller.unmarshal(reader); 


       return input.getPath();  
      } 
     }); 

ここで、input.getPath（）は、ファイルのhdfsパスです。

JavaDstreamオブジェクトを収集するオプションはありません。それ以外の場合は、最初にデータを収集し、データをファイルから取得することで使用します。

Iamは新しいRDD内部マップ関数を作成できません。タスクはシリアル化できません。

他にもオプションはありますか？

Answer 1

foreachRDDを使用できます。 RDDSはドライバのみで作成することができます - それは、ドライバ上で実行されるので、RDDアクションが

transformed.foreachRDD (rdd -> { 
    String inputPath = doSomethingWithRDD(rdd) 
    rdd.sparkContext.textFile(inputPath) ... 
}); 

はあなたが変換またはアクション内のRDDを作成することができないことに注意してください許可されています。 foreachRDDの例と同様の質問はhereです。これは、マップ、フィルタまたはforeachPartition内のSparkContextを使用できないことを意味します