如何在Spark中使用java平行处理每行文件？

Question

我有一个包含多个网址的文件。 我想阅读每个这个网址，并对其进行一些处理。 由于处理部分是独立的每个网址，我想在Spark上并行执行此操作。

SparkConf conf = new SparkConf().setAppName("org.sparkexample.WordCount").setMaster("local"); 
JavaSparkContext context = new JavaSparkContext(conf); 
JavaRDD<String> textFile = context.textFile("urlFile"); 
/* Now for each line of this textFile I need to call below */ 

ExtractTrainingData ed = new ExtractTrainingData(); 
List<Elements> list = ed.getElementList(inputUrl); 
ed.processElementList(inputUrl, list); 

任何人都可以建议我该怎么做？

Answer 1

如果每个URL是在其他行，那么你可以做的foreach：

SparkConf conf = new SparkConf().setAppName("org.sparkexample.WordCount").setMaster("local"); 
JavaSparkContext context = new JavaSparkContext(conf); 
JavaRDD<String> textFile = context.textFile("urlFile"); 

textFile.foreach (new VoidFunction<String>() { 
    public void call (String line) { 
     // this code will be executed parallely for each line in file 
     ExtractTrainingData ed = new ExtractTrainingData(); 
     List<Elements> list = ed.getElementList(inputUrl); 
     ed.processElementList(inputUrl, list); 
    } 
}); 

如果结果列表也应并行，则：

SparkConf conf = new SparkConf().setAppName("org.sparkexample.WordCount").setMaster("local"); 
JavaSparkContext context = new JavaSparkContext(conf); 
JavaRDD<String> textFile = context.textFile("urlFile"); 

textFile.map (new Function<String, List<Elements>() { 
    public List<Elements> call (String line) { 
     // this code will be executed parallely for each line in file 
     ExtractTrainingData ed = new ExtractTrainingData(); 
     List<Elements> list = ed.getElementList(inputUrl); 
     return list; 
    } 
}).flatMap (list -> list.iterator()) 
.foreach ((String element) -> { 
    // here put code that is in processElementList 
}); 

我用lambda语法，你可以使用，当然匿名函数

编辑：确保Elements是序列化