Apache Spark - partitionについて -

#<i class="fa fa-book" style="font-size:1em;"></i>はじめに
この記事は、`Apache Spark`の`partition`の概念について例題を添えてまとめてみようと試みた記事です。
`Apache Spark`の概要については[Apache Spark で分散処理入門](http://qiita.com/Hiroki11x/items/4f5129094da4c91955bc)をどうぞ。


#<i class="fa fa-folder-o" aria-hidden="true"></i>例題で見るパーティションのイメージ
以下は、`Apache Spark`の分散処理のイメージを例題とともに図解しているものです。
まず、使う関数の説明、次にその処理のイメージ、最後にソースコードの順番で紹介しています。
パーティションの使い方が異なる三例を紹介します。
- filter関数の例
- flatmap関数の例
- reduceByKey関数の例

##__<i class="fa fa-code" aria-hidden="true"></i>filter関数の例__

要素単位の変換としては`map()`や`filter`などがあります。

| 関数 | 説明 |
|:-----------|:------------|
| map()  | 引数に関数を一つ取り、その関数をRDD内の各要素に適応し、その結果を新しい値とするRDDを返す |
| filter()  | 引数に関数を一つ取り、そのフィルタ関数が真になる要素だけを含むRDDを返す |

![スクリーンショット 2016-10-11 18.15.57.png](https://qiita-image-store.s3.amazonaws.com/0/56510/e5083b31-d94e-e61e-6583-a5d34e18881d.png)

![スクリーンショット 2016-10-11 18.16.04.png](https://qiita-image-store.s3.amazonaws.com/0/56510/a42c315f-3d54-9c0f-83e0-4b1f1c8dd8de.png)

![スクリーンショット 2016-10-11 18.16.07.png](https://qiita-image-store.s3.amazonaws.com/0/56510/1fd57063-75df-7e64-60d7-acd48e969faf.png)


ソースコードはこちら

```Sample02.java
package other;

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

/**
 * Created by hirokinaganuma on 2016/10/11.
 */
public class Sample02 {
    public static void main(String[] args) throws Exception {
        String master;
        if (args.length > 0) {
            master = args[0];
        } else {
            master = "local";
        }
        JavaSparkContext sc = new JavaSparkContext(master, "basicavg", System.getenv("SPARK_HOME"), System.getenv("JARS"));
        JavaRDD<String> input = sc.textFile("bin/input/*.txt");
        JavaRDD<String> result = input.filter(s-> !s.contains("Triangle"));
        result.saveAsTextFile("bin/output/output02");
        sc.stop();
    }
}
```

```filter/sample01.txt
Rectangle
Circle
Triangle
```

```filter/sample02.txt
Circle
Rectangle
Triangle
```

```filter/sample03.txt
Triangle
Rectangle
```

`RDD`は大量のデータを要素として保持する分散コレクションです。`RDD`は複数のマシンから構成されるクラスタ上での分散処理を前提として設計されており、内部的には__`partition`__という塊に分割されています。`Spark`ではこの`partition`が分散処理の単位となっています。`RDD`を`partition`ごとに複数のマシンで処理することによって、単一のマシンでは処理できないデータを扱うことができます。

このように分割されたファイルを一つの`RDD`として扱い、`input.filter(s-> !s.contains("Triangle"))`と記述するだけで、分散を意識せずプログラミングできるようになってます。

##__<i class="fa fa-code" aria-hidden="true"></i>flatmap関数の例__
`filter`や`map`関数が1:1の関係だったのに対し、この関数は、一つの要素から複数の要素を生成する1:Nの関係の変換です。

| 関数 | 説明 |
|:-----------|:------------|
| flatMap()  | 引数に関数を一つ取り、その関数をRDD内の各要素に適応して呼ばれるが、この関数はその結果を返すIteratorを返します。それらのIterator全てから返された要素を値とするRDDを最優的に返します |

![スクリーンショット 2016-10-11 20.09.53.png](https://qiita-image-store.s3.amazonaws.com/0/56510/b1b71477-4658-5577-f172-77af4e232549.png)

![スクリーンショット 2016-10-11 20.09.55.png](https://qiita-image-store.s3.amazonaws.com/0/56510/b72df8f9-617c-fdf0-82ab-208b571fc4d5.png)

![スクリーンショット 2016-10-11 20.09.57.png](https://qiita-image-store.s3.amazonaws.com/0/56510/e7054334-70ec-24ee-2363-790d9087c67b.png)

ソースコードはこちら

```Sample03.java
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;

import java.util.Arrays;

/**
 * Created by hirokinaganuma on 2016/10/11.
 */
public class Sample03 {
    public static void main(String[] args) throws Exception {
        String master;
        if (args.length > 0) {
            master = args[0];
        } else {
            master = "local";
        }
        JavaSparkContext sc = new JavaSparkContext(master, "basicavg", System.getenv("SPARK_HOME"), System.getenv("JARS"));
        JavaRDD<String> input = sc.textFile("bin/input/flatmap/*.txt");
        JavaRDD<String> result = input.flatMap(
                new FlatMapFunction<String, String>() {
                    public Iterable<String> call(String x) {
                        return Arrays.asList(x.split(" "));
                    }
                }
        );
        result.saveAsTextFile("bin/output/output03");
        sc.stop();
    }
}
```

```flatmap/sample01.txt
I am from Fukuoka
Hello Apache Spark
```

```flatmap/sample02.txt
Tokyo Institute of Technology
```

##__<i class="fa fa-code" aria-hidden="true"></i>reduceByKey関数の例__
`reduceByKey`は返還前の`RDD`に含まれる要素を、同じ`RDD`に含まれるほかの`partition`の要素とまとめて処理する必要のあるものです。この変換はkey・valueペアを要素とする`RDD`を対象にしており、同じkeyを持つ要素をまとめて処理します。`Spark`は`partition`ごとに独立して分散処理を行うため同じkeyを持つ要素はすべて同じ`partition`に含まれている必要があります。そのため`reduce`処理の前に配置換え(`shuffle`)が行われます。`shuffle`は、返還前の`RDD`の要素を、keyに基づいて変換後の`RDD`の`partition`に振り分けるように行われます。そのため、`shuffle`によって同じkeyを持つ要素が同じ`partition`に含まれることが保証されます。
下の図では表しきれませんでしたが、`shuffle`の前にそのpartition`ないで`reduce`処理が行われ、通信コストを減らしているようです。
https://databricks.gitbooks.io/databricks-spark-knowledge-base/content/best_practices/prefer_reducebykey_over_groupbykey.html

| 関数 | 説明 |
|:-----------|:------------|
| reduceByKey()  | 引数に関数を一つ取り、その関数は同じキーの値を結合する際のreduce処理である |

![スクリーンショット 2016-10-11 21.32.57.png](https://qiita-image-store.s3.amazonaws.com/0/56510/5cea486b-cf28-7fb7-ad6f-3eea8b70b3b5.png)

![スクリーンショット 2016-10-11 21.33.05.png](https://qiita-image-store.s3.amazonaws.com/0/56510/42c29f9f-44d1-ef9b-c610-aee0616593d8.png)

![スクリーンショット 2016-10-11 21.33.07.png](https://qiita-image-store.s3.amazonaws.com/0/56510/a51e8f61-bd48-c274-47fb-516f16336f55.png)

ソースコードはこちら

```Sample04.java
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;

/**
 * Created by hirokinaganuma on 2016/10/11.
 */
public class Sample04 {
    public static void main(String[] args) throws Exception {
        String master;
        if (args.length > 0) {
            master = args[0];
        } else {
            master = "local";
        }
        JavaSparkContext sc = new JavaSparkContext(master, "basicavg", System.getenv("SPARK_HOME"), System.getenv("JARS"));
        JavaRDD<String> input = sc.textFile("bin/input/reducebykey/*.txt");
        JavaRDD<String> words = input.flatMap(
                new FlatMapFunction<String, String>() {
                    public Iterable<String> call(String x) {
                        return Arrays.asList(x.split(" "));
                    }
                }
        );

        JavaPairRDD<String, Integer> pairrdd = words.mapToPair(
                new PairFunction<String, String, Integer>(){
                    public Tuple2<String, Integer> call(String x){
                        return new Tuple2(x, 1);
                    }
                }
        );

        JavaPairRDD<String, Integer> output = pairrdd.reduceByKey(new Function2<Integer, Integer, Integer>(){
            public Integer call(Integer x, Integer y){
                return x+y;
            }
        });

        output.saveAsTextFile("bin/output/output04");
    }
}
```


```reducebykey/sample01.txt
fish
cat
```

```reducebykey/sample02.txt
cat
fish
```

```reducebykey/sample03.txt
fish
```

```reducebykey/sample04.txt
cat
cat
dog
```



#<i class="fa fa-question-circle-o" aria-hidden="true"></i>`Apache Spark`の`partition`とは何なのか

`RDD`は大量のデータを要素として保持する分散コレクションです。`RDD`は複数のマシンから構成されるクラスタ上での分散処理を前提として設計されており、内部的には__`partition`__という塊に分割されています。`Spark`ではこの`partition`が分散処理の単位となっています。`RDD`を`partition`ごとに複数のマシンで処理することによって、単一のマシンでは処理できないデータを扱うことができます。


##<i class="fa fa-heart" aria-hidden="true"></i>モチベーション
分散プログラムでは通信は非常にコストが高く、ネットワークのトラフィックを最低限に抑えることはパフォーマンスの大幅な改善につながります。
`Spark`のプログラムは`RDD`のパーティショニングを制御して通信を削減していく必要があります。

例えば`Spark`はkeyの集合がまとまってあるノードに現れることをプログラムで保証することができます(reduceByKey()で紹介しました)。特にパーティショニングが役立つのはデータセットが結合のようなキー操作を複数回再利用される場合に限ります。というのも`Spark`の操作の多くはネットワーク越しにキーに基づくシャッフルを起こします。それらはすべてパーティショニングの恩恵を受けることができます。

##<i class="fa fa-sitemap" aria-hidden="true"></i>データのパーティショニング
`Java`や`Scala`では`partitioner`プロパティを用いて`RDD`のパーティショニングの方法を指定できます。ここで`spark.Partitioner`オブジェクトの値として、`RDDに対してそれぞれのキーの行き先のパーティションを知らせるものをセットします。
`reduceByKey()`のように単一の`RDD`に対して働く操作をパーティション化された`RDD`上で動作させると、各keyのすべての値は、単一マシン上でローカルに計算され、ローカルに`reduce()`された最終の値だけが、各ワーカーノードからマスターに戻されます(先ほど紹介しました)。
より細かく説明すると、シャッフル時に同じkeyを持つ要素を同じ`partition`に振り分けるのは`partitioner`の仕事で、`partitioner`は変換後の`RDD`の`partition`数と振り分け対象の要素のkeyの内容をもとに、要素を振り分ける先の`partition`を決定します。`Spark`ではデフォルトで`key`のハッシュ値を変換後の`RDD`の`partition`数で割ったあまりをもとに、振り分け先の`partition`を決定します。

また、二つの`RDD`に対して操作する場合は、事前にパーティショニングが行われていると、最低でも二つの`RDD`のうちの片方はシャッフルされません。
さらに、`Spark`はそれぞれの操作がパーティショニングに及ぼす内部的な影響を知っているので、データパーテショニングを行う操作によって生成される`RDD`に、自動的に`partitioner`を設定します。

##<i class="fa fa-hdd-o" aria-hidden="true"></i>永続化
補足ですが`RDD`に対してパーティショニングを行い、それによって生成された`RDD`を変換の対象とするのであれば`persist()`で永続化すべきです。理由としては、それ以降の`RDD`のアクションはパーティショニングで生成した`RDD`の系統全体を評価し直すことになるので、パーティショニング前の`RDD`に何度もハッシュパーティショニングが行われてしまい意味がなくなってしまいます。

#<i class="fa fa-book" style="font-size:1em;"></i>参考
- http://www.ne.jp/asahi/hishidama/home/tech/scala/spark/partition.html
- https://jaceklaskowski.gitbooks.io/mastering-apache-spark/content/
- http://qiita.com/Hiroki11x/items/4f5129094da4c91955bc