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

注：ファイルのパスは、既存のパスであってはいけません。つまり、ファイルは存在してはいけません。

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

それが開始された意味：あなたはこのコードを実行する場合は、プログラムに次の行を追加して実行した場合、何もexpected.Howeverとして起こりません

は

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

それは次の例外を提供しますDAGを実行します。 reduceByKey（）は変換であるため、collect（）やtake（）などのアクションが実行されるまではそうでないはずです。

スパークバージョン：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 FileSystemへの単一の呼び出しは複数のファイルを返すことができます。 InputFormatで定義されたさまざまな最適化に従って、実際にそれらを探し出すことによってのみ知ることができます）。

実際のDAGではなく、実際のDAG（ここではmap/flatMap/aggregateなど）がここで実行されています。

おそらくキーバリアントによって減らすこの中で、あなた自身のpartitionnerを提供することで、それを避けることができます：

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