Apache Sparkでコール・オプション価格の分散処理計算 with Google Cloud Dataproc

モンテカルロ法でコール・オプション価格 - My Life as a Mock Quantを流行りの分散処理環境Apache Sparkでやってみたという話。
コード書くのは一瞬なんだけど、依存関係周りの解消が辛かったので、積極的にコード残しておきたい。

下記コード+sbtファイルを用意してコンソールから、バイナリ作成、クラスタ起動、クラスタへジョブ投入が

sbt assembly
gcloud beta dataproc clusters create dataproc01
gcloud beta dataproc jobs submit spark --cluster dataproc01 --class App --jars spark-optionpricing-assembly-1.0.jar

なかんじでいける。計算終了後は

gcloud beta dataproc clusters delete dataproc01

で破産しないようにクラスタを殺す。

コード

プロジェクト全体

https://github.com/teramonagi/spark-examples/tree/master/spark-optionpricing

Scala部分のみ

import org.apache.spark.mllib.random.StandardNormalGenerator
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import scala.math

object App {
  def main(args: Array[String]) = {
    val random = new StandardNormalGenerator()
    // Monte Carlo setting
    val size = math.pow(10, 10).toInt
    // market information
    val volatility = 0.3
    val spotPrice = 100.0
    val rate = 0.01
    // option parameters
    val maturity = 5.0
    val strike = 102.0

    // Spart setting
    // For local use
    //val conf = new SparkConf().setAppName("Estimating Call option price").setMaster("local[*]")
    val conf = new SparkConf().setAppName("Estimating Call option price")
    val sc = new SparkContext(conf)

    //Monte Cralo
    val result = sc.parallelize(1 to size).map { i =>
      val noize =  volatility * math.sqrt(maturity) * random.nextValue()
      val drift = (rate - 0.5 * volatility * volatility) *  maturity
      val price = spotPrice * math.exp( drift + + noize )
      math.max(price - strike, 0.0)
    }.sum * ( math.exp( -rate * maturity) )/ size
    // Show result
    println("Option price is: " + result)
    sc.stop()
  }
}

コード

参考