通過Spark分別逐個處理多個文件

Question

我需要使用Apache Spark實現一個工作流的幫助。我的任務在下：

1. 我有幾個CSV文件作爲源數據。注意：這些文件可能有不同的佈局
2. 我有元數據信息我如何需要解析每個文件（這不是問題）
3. 主要目標：結果是包含多個附加列的源文件。我必須更新每個源文件而不加入一個輸出範圍。例如：源10個文件 - > 10個結果文件，每個結果文件只有來自相應源文件的數據。

據我所知星火可以通過面具打開多個文件：

var source = sc.textFile("/source/data*.gz"); 

但在這種情況下，我不能識別文件的哪一行。如果我得到的源文件的列表，並嘗試過程以下情形：

JavaSparkContext sc = new JavaSparkContext(...); 
List<String> files = new ArrayList() //list of source files full name's 
for(String f : files) 
{ 
    JavaRDD<String> data = sc.textFile(f); 
    //process this file with Spark 
    outRdd.coalesce(1, true).saveAsTextFile(f + "_out"); 
} 

但在這種情況下，我會處理在連續模式下的所有文件。

我的問題是下一個：我怎麼可以在並行模式下處理很多文件？例如：一個文件 - 一個執行者？

非常感謝您的幫助！

Answer 1

下面是步驟

1. 使用sparkcontext.wholeTextFiles（「/路徑/到/文件夾/含有/所有/文件」）
2. 上述返回一個RDD其中鍵是文件的路徑和值是文件的內容
3. rdd.map（拉姆達X：X [1]） - 這給你只文件內容的RDD
4. rdd.map（拉姆達X：customeFunctionToProcessFileContent（X））
5. 由於地圖功能並行工作，你做的操作會更快，而且沒有順序 - 只要你的任務不依賴於彼此，這是並行的主要標準。

上面的工作與默認分區雖然。所以你可能得不到輸入文件計數等於輸出文件數（因爲輸出是分區數）。

根據您的數據，您可以根據計數或任何其他唯一值重新分區RDD，因此最終輸出文件計數等於輸入計數。這種方法將只有並​​行性，但不會達到最佳分區數目所達到的性能。

Answer 2

您可以打開常規java固定大小的線程池（比如說10個線程），並從Callable/Runnable提交sparkAssertFile。 這將提交10個並行作業，並且如果您的火花簇中有足夠的資源 - 它們將並行執行。 類似如下

import java.util.ArrayList; 
import java.util.List; 
import java.util.concurrent.Executor; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 
import java.util.concurrent.Future; 

import org.apache.spark.api.java.JavaRDD; 
import org.apache.spark.api.java.JavaSparkContext; 

import com.google.common.collect.Lists; 

public class Test { 

    public static void main(String[] argv) { 
     final JavaSparkContext sc = new JavaSparkContext(...); 
     List<String> files = new ArrayList<>(); //list of source files full name's 
     ExecutorService pool = Executors.newFixedThreadPool(10); 
     List<Future<?>> futures = new ArrayList<>(); 
     for(final String f : files) 
     { 
      Future<?> fut = pool.submit(new Runnable() { 

       @Override 
       public void run() { 
        JavaRDD<String> data = sc.textFile(f); 
        //process this file with Spark 
        outRdd.coalesce(1, true).saveAsTextFile(f + "_out"); 

       } 
      }); 
      futures.add(fut); 

     } 
     //waiting for all tasks 
     for (Future<?> fut : futures) { 
      fut.get(); 
     } 
    } 
}