Apache Spark：爲什麼reduceByKey轉換執行DAG？

Question

我正面臨一個奇怪的問題。據我瞭解，Spark中的DAG操作只在執行操作時執行。但是，我可以看到reduceByKey（）操作（作爲轉換）開始執行DAG。

重現步驟。嘗試以下代碼段

SparkConf conf =new SparkConf().setMaster("local").setAppName("Test"); 
JavaSparkContext context=new JavaSparkContext(conf); 

JavaRDD<String> textFile = context.textFile("any non-existing path"); // This path should not exist 

JavaRDD<String> flatMap = textFile.flatMap(x -> Arrays.asList(x.split(" ")).iterator()); 
JavaPairRDD<String, Integer> mapToPair = flatMap.mapToPair(x -> new Tuple2<String, Integer>((String) x, 1)); 

注意：文件的路徑不應該是任何現有的路徑。換句話說，文件不應該存在。

如果執行此代碼，沒有什麼如果你添加下面一行到程序和執行情況作爲expected.However

mapToPair.reduceByKey((x, y) -> x + y); 

它提供了以下異常：

Exception in thread "main" org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: 

這意味着它已經開始執行DAG。由於reduceByKey（）是一種轉換，因此在執行諸如collect（）或take（）之類的操作之前，情況不應該如此。

Spark版本：2.0.0。請提供您的建議。

Answer 1

這是因爲它並不是真正的DAG被執行（例如：它的整個實現）。

發生什麼是reduceByKey 需要分區程序才能工作。如果你不提供一個，Spark會根據約定和默認值創建一個。 「默認partiionner」作爲代碼如下評論：

/** 
* Choose a partitioner to use for a cogroup-like operation between a number of RDDs. 
* 
* If any of the RDDs already has a partitioner, choose that one. 
* 
* Otherwise, we use a default HashPartitioner. For the number of partitions, if 
* spark.default.parallelism is set, then we'll use the value from SparkContext 
* defaultParallelism, otherwise we'll use the max number of upstream partitions. 
* 
* Unless spark.default.parallelism is set, the number of partitions will be the 
* same as the number of partitions in the largest upstream RDD, as this should 
* be least likely to cause out-of-memory errors. 
* 
* We use two method parameters (rdd, others) to enforce callers passing at least 1 RDD. 
*/ 

這一定義意味着，在某些情況下，計算分區的所有上游RDDS數量。在你的情況下，這意味着要求「文件系統」（可以是Hadoop，可以是本地的......）執行任何必要的操作（例如，對Hadoop文件系統的單個調用可以返回多個文件，每個文件也可以被分割根據其InputFormat定義的各種優化，只能通過實際查找它們才能知道）。

這就是在這裏執行的，而不是實際的DAG（例如，你的map/flatMap/aggregate，...）。

您可以通過在此者皆變種減少提供自己partitionner可能避免呢？

reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]