Spark分别处理多个文件

Question

我在使用 Apache Spark 实施一个工作流方面需要帮助。我的任务在下一个：

1. 我有几个 CSV 文件作为源数据。注意：这些文件可能有不同的布局
2. 我有元数据，其中包含我需要如何解析每个文件的信息（这不是问题）
3. 主要目标：结果是带有几个附加列的源文件。我必须更新每个源文件而不加入一个输出范围。例如：源 10 个文件 -> 10 个结果文件，每个结果文件只有对应源文件的数据。

据我所知，Spark 可以通过掩码打开许多文件：

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("/path/to/folder/ contains/all/files")
2. 上面返回一个RDD，其中key是文件的路径，value是文件的内容
3. rdd.map(lambda x:x[1]) - 这给你一个只有文件内容的rdd
4. rdd.map(lambda x: customeFunctionToProcessFileContent(x))
5. 由于 map 函数是并行工作的，因此您执行的任何操作都会更快且不连续 - 只要您的任务不相互依赖，这是并行性的主要标准

上述方法适用于默认分区。所以你可能不会得到输入文件数等于输出文件数（因为输出是分区数）。

您可以根据计数或基于您的数据的任何其他唯一值对 RDD 重新分区，因此最终输出文件计数等于输入计数。这种方法仅具有并行性，但不会达到最佳分区数的性能

Answer 2

您可以打开常规 java 固定大小的线程池（例如 10 个线程）并从 Callable/Runnable 提交您的 saveAsTextFile 的 spark 作业。 这将提交 10 个并行作业，如果您的 spark 集群中有足够的资源 - 它们将并行执行。 类似以下的东西

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();
        }
    }
}