S-JIS[2014-09-22/2017-01-18] 変更履歴

Sparkアプリケーション実行方法

Apache Sparkアプリケーションの実行方法のメモ。

概要

Sparkアプリケーションを実行するにはspark-submitコマンドを使用する。
アプリケーションはコンパイルしてjarファイルにしておく必要がある。

例

Linux上でSparkアプリケーションを実行してみる例。

RDDサンプルのソースは以下のような感じ。

src/example/Example.scala：

package example

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

case class MyData(var n: Int)

object Example {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("example")
    val sc = new SparkContext(conf)
    try {
      printf("master=%s, name=%s\n", sc.master, sc.appName)

      val rdd1 = sc.makeRDD(Seq(new MyData(123), new MyData(456), new MyData(789)))
      val rdd2 = rdd1.map { a => a.n += 1; a }
      rdd2.collect.foreach(println)
      rdd1.collect.foreach(println)

    } finally {
      sc.stop()
    }
  }
}

このサンプルのポイントは、アプリケーション名だけ指定し、マスターURLは指定していないこと。
マスターURLはデフォルト値を使う。つまり実行時に外部から指定する想定。

アプリケーション名も実行時に外部から指定できるが、普通は（環境によって変えたりしないのでない限り）固定だと思うので、このサンプルではプログラム内で指定した。

まず、ソースをコンパイルしてjarファイルを作成する。

$ ls -R src
src:
example

src/example:
Example.scala

$ scalac -cp $($SPARK_HOME/bin/compute-classpath.sh) -d example.jar src/example/Example.scala

$ ls
example.jar  src

scalacは、普通のScalaのコンパイルコマンド。
Spark用のクラスパス（jarファイル群）は、compute-classpath.shというシェルを実行すると得られるので便利。

spark-submitコマンドを使って実行する。

$ $SPARK_HOME/bin/spark-submit --master "local[2]" --class example.Example example.jar

マスターURLは「--master」オプション、実行するクラス名（オブジェクト名）は「--class」オプションで指定する。
（「--name」オプションでアプリケーション名も指定できるが、今回のサンプルではプログラム内でアプリケーション名をセットしているので、このコマンドで指定しても無視される）
アプリケーションが入っているjarファイルは（オプション名なしで）そのまま指定する。

起動オプション

spark-submitの起動オプション。

その他に、Spark standaloneやMesos・YARN固有のオプションが存在する。

SparkUI

Sparkアプリケーションを実行するとSparkUIというウェブサービスが実行されるので、ブラウザーでSparkアプリケーションの実行状況を見ることが出来る。

Sparkアプリケーションを実行すると、起動ログに以下のようなログが出ている。

14/09/22 11:44:18 INFO SparkUI: Started SparkUI at http://hishidama-hadoop:4040

このログの「hishidama-hadoop」というのは、実行しているマシンに付けているサーバー名。
ブラウザーで「http://hishidama-hadoop:4040」を指定すれば、SparkUIにアクセスできる。

Sparkアプリケーションが終了するとSparkUIにもアクセスできなくなる。

ローカルで実行する方法

マスターURLにlocalを指定すると、ローカル（スタンドアローン）で実行することが出来る。[2017-01-18]

ローカルで実行する場合は、ドライバーと同一のJavaVM上で（マルチスレッドで）executorが実行される。
したがってexecutor用のJavaVMの設定は意味が無く、何か設定したい場合はドライバーに対して行う。
Spark2.1.0の場合、ドライバーホストをlocalhostにする必要がある。（参考：y_t_pさんのSpark-shell起動時のエラー）

$ spark-submit --master 'local[*]' --conf spark.driver.host=localhost --driver-memory 2g --driver-java-options "-Djava.io.tmpdir=$HOME/spark-submit-tmp" --class クラス名 jarファイル名 アプリケーション引数

設定方法は、Sparkの設定ファイル（SPARK_HOME/conf/spark-defaults.conf）に記述する方法、環境変数で指定する方法、spark-submitの引数に指定する方法がある。
環境変数で指定する場合は、SPARK_HOME/conf/spark-env.shに記述しておくと、実行時に読み込まれる。

• どのような環境変数が使えるのかは、Spark Standalone Modeを参照（一部の環境変数は載っていないが…）
• 設定の解釈・構築はorg.apache.spark.launcher.Mainで行っている。具体的にはSparkSubmitCommandBuilderで 構築している（spark-submit以外のコマンドの場合はSparkClassCommandBuilder）。
• 中間ファイルを出力する場所は、デフォルトでは/tmpの下。なので/tmpの容量が少ないとエラーになることがある。java.io.tmpdirを指定すると、その場所に出力されるようになる。

Spark2.1.0では、localで動かす場合にドライバーホストを設定していないと、以下のようなエラーが発生する。

17/01/18 21:46:58 ERROR SparkContext: Error initializing SparkContext.
java.net.BindException: 要求アドレスに割り当てられません: Service 'sparkDriver' failed after 16 retries (starting from 0)! Consider explicitly setting the appropriate port for the service 'sparkDriver' (for example spark.ui.port for SparkUI) to an available port or increasing spark.port.maxRetries.
        at sun.nio.ch.Net.bind0(Native Method)
        at sun.nio.ch.Net.bind(Net.java:433)
        at sun.nio.ch.Net.bind(Net.java:425)
        at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:223)
        at io.netty.channel.socket.nio.NioServerSocketChannel.doBind(NioServerSocketChannel.java:127)
        at io.netty.channel.AbstractChannel$AbstractUnsafe.bind(AbstractChannel.java:501)
        at io.netty.channel.DefaultChannelPipeline$HeadContext.bind(DefaultChannelPipeline.java:1218)
        at io.netty.channel.AbstractChannelHandlerContext.invokeBind(AbstractChannelHandlerContext.java:506)
        at io.netty.channel.AbstractChannelHandlerContext.bind(AbstractChannelHandlerContext.java:491)
        at io.netty.channel.DefaultChannelPipeline.bind(DefaultChannelPipeline.java:965)
        at io.netty.channel.AbstractChannel.bind(AbstractChannel.java:210)
        at io.netty.bootstrap.AbstractBootstrap$2.run(AbstractBootstrap.java:353)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:408)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:455)
        at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:140)
        at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
        at java.lang.Thread.run(Thread.java:745)