S-JIS[2017-01-14] 変更履歴

SparkSession

Apache Sparkで最初に使うSparkSessionクラスについて。

概要

Dataset（Spark1.6以降）（あるいはDataFrame（Spark1.3〜1.5））を使う為には、最初にSparkSessionインスタンスを生成する必要がある。

import org.apache.spark.sql.SparkSession

パッケージがspark.sqlなのがちょっと気になるけど＾＾；、元々SQLを扱う目的で始まったらしい。

SparkSessionの生成

SparkSessionインスタンスの生成には、SparkSessionのビルダーを使用する。

    val spark = SparkSession.builder().getOrCreate()

（SparkSessionの内部ではSparkContext（やSparkConf）が生成されて保持される）

この場合、マスターURLやアプリケーション名はspark-submitによって指定されたものが使われる。

マスターURLやアプリケーション名等をプログラム内で指定するには以下のようにする。

    val spark = SparkSession.builder()
      .master("local")
      .appName("example")
      .config("キー", "値")
      .getOrCreate()

SparkContextに何かセットしたい場合は、SparkSessionからSparkContextを取得する。

    val sc = spark.sparkContext
    sc.setCheckpointDir("/tmp/spark/checkpoint")

SparkSessionインスタンスを生成すると、HTTPサーバーとしての機能が起動する。
ので、終了するときはstopメソッドを呼び出す。

    val spark = 〜
    try {
      〜
    } finally {
      spark.stop()
    }

stopすることで、各executorも終了する。

このstopメソッドは、内部でSparkContextのstopを呼び出している。
Spark2.1ではcloseメソッドもあるが、stopを呼び出しているだけ。

SparkSessionの暗黙変換メソッド

SparkSessionを利用する暗黙変換メソッドが色々あるので、インポートしておくと便利。

     val spark = SparkSession.builder().〜.getOrCreate()
     import spark.implicits._

Scalaでは、インスタンス内のobjectのメソッドをインポートすることが出来る。上記の「spark.implicits」は、spark変数（で保持されているインスタンス）のimplicitsオブジェクトを指している。

主に以下のことが出来るようになる。

• ケースクラス・数値・String・配列用のEncoderの取得。
• RDDやSeqからDatasetへの変換。
  • Seqに対してtoDS()（やtoDF()）が出来るようになる。

implicitsオブジェクトはSpark2.0.0ではSQLImplicitsクラスを継承しており、実際の暗黙変換メソッドはそちらで定義されている。

class SparkSession {
〜
  object implicits extends SQLImplicits with Serializable {
    protected override def _sqlContext: SQLContext = SparkSession.this.sqlContext
  }
〜
}

SparkSessionのメソッド

SparkSessionの主なメソッド。

• 設定系
• 実行系
• DataFrame系
• Dataset系

設定系

実行系

DataFrame系

Dataset系