JSONライブラリ性能比較

Spark SQL 1.3の登場を機にバッチ処理基盤の刷新を考えています。この流れの中でJobSchedulerやSpark SQLをDockerで動かす試み(Docker ComposeでMySQLを使う,DockerでSpark SQL)などを行ってきました。

バッチをSpark SQLで記述し、データや計算量の規模に応じてDocker Cluster(e.g. ECS)またはSpark Cluster(e.g. EMR)を選択してバッチ処理を実行するという枠組みが見えてきました。

次に考えておきたいのはバッチ処理で使用する要素技術の選択です。今回はJSONライブラリについて性能の観点から味見してみました。

なお、あくまでも味見レベルの測定なので、条件を変えると違った結果になる可能性も高いです。また、ありがちですが性能測定プログラムにバグがあって結果が逆にでるようなことがあるかもしれません。このようなリスクがあることを前提に参考にして頂ければと思います。

オンライン処理とバッチ処理

クラウド・アプリケーションはおおまかにいうとWebページやREST APIとして実現するフロント系(オンライン系)と、バックエンドで動作するバッチ系で構成されます。(今後はこれに加えてストリーム系が入ってきます。またバッチ系とはニュアンスの異なる分析系も別立てで考えておくとよさそうです。)

極簡単なアプリケーションではフロント系だけで成立する場合もありますが、ある程度本格的な機能を提供する場合はバッチ系は必須になってきます。また、クラウド・アプリケーション・アーキテクチャの定番であるCQRS+Event Sourcingはバッチ系が主役ともいえるアーキテクチャであり、今後バッチ系の重要度はますます高まってくることは間違いないでしょう。

オンライン系のロジックとバッチ系のロジックでは以下のような特性の違いがあります。

	オンライン処理	バッチ処理
性能特性	レスポンス重視	スループット重視
データ規模	小	大
計算量	小	大
実行時間	小	大
エラー処理	即時通知	ジョブ管理

こうやって整理すると様々な面で求められる特性が大きく異なることが分かります。

これらの特性を念頭にプログラミングしていくことは当然ですが、使用する要素技術やライブラリもこれらの特性に適合したものを選択することが望ましいです。

オンライン処理はデータ規模や計算量が小さいので使いやすさ重視でよいですが、バッチ処理はデータ規模や計算量が大きいので処理性能やメモリ消費量がクリティカルな要因になります。多少使い難くても、高速、メモリ消費量が少ない、ものを選ぶ方がよいでしょう。

バッチ処理の中のJSON

バッチ処理の中でのJSONの利用シーンですが、主に意識しなければいけないのは大規模JSONファイルの入力ということになるかと思います。ログファイルや移入データがJSON形式で用意されているようなケースですね。

通常この用途のデータはCSVやLTSVで用意されるケースが多いと思いますが、データが木構造になっている場合(いわゆる文書型)にはJSONやXMLデータが適しており、JSONで用意されるケースも多いでしょう。

バッチをSpark SQLで組む場合、データがCSVで用意されていたり、JSONであってもフラットな構造の場合はSpark SQLで簡単に取り込めるのでJSONライブラリの出番はありません。

ということで、ある程度複雑な構造を持ったJSONファイルがメインの操作対象となります。

ただ、バッチをSpark SQLではなく通常のScalaプログラムとして組みたいケースも出てくるはずなので、フラットな構造のJSONファイルに対してもペナルティなしで扱える必要はあります。

ユースケース1: 複雑な構造を持ったJSON→CSV

代表的なユースケースとしては、複雑な構造を持った大規模JSONファイルから必要な項目を抽出してCSVに書き出す、といった用途が考えられます。

SparkやHadoopの前処理として頻出しそうなユースケースです。

このケースでは(1)JSONの基本的なパース性能と、(2)JSONからデータを抽出したりデータのフィルタリングをしたりする処理の記述性が論点となります。

ユースケース2: 複雑な構造を持ったJSON→アプリケーションロジック適用

少し複雑なバッチ処理を行う場合は、アプリケーションロジックを適用するためcase classなどで定義したドメイン・オブジェクトにJSONを変換する必要がでてきます。

case classに変換後は、アプリケーションが管理しているドメインロジックを型安全に適用することができるので、プログラムの品質と開発効率を高めることができます。

ただ、JSONをcase classにマッピングする処理はそれなりに重たいので、JSON段階である程度フィルタリングした上で、必要なデータのみcase classにマッピングする形を取りたいところです。

テストプログラム

検証対象のJSONライブラリは以下の4つにしました。

• Json4s native
• Json4s jackson
• Play-json
• Argonaut

各ライブラリの性能測定対象のプログラムは以下になります。

どのライブラリもほとんど使い方は同じです。ただし、ArgonautのみJSONとオブジェクトのマッピング設定がかなり煩雑になっています。

Json4s native

view plainprint?

1. package sample  
2.   
3. import org.json4s._  
4. import org.json4s.native.JsonMethods._  
5.   
6. object Json4sNativeSample {  
7.   implicit val formats = DefaultFormats  
8.   
9.   def jsonSimple() = {  
10.     val s = Company.example  
11.     parse(s)  
12.   }  
13.   
14.   def jsonComplex() = {  
15.     val s = Person.example  
16.     parse(s)  
17.   }  
18.   
19.   def simple() = {  
20.     val s = Company.example  
21.     val j = parse(s)  
22.     j.extract[Company]  
23.   }  
24.   
25.   def complex() = {  
26.     val s = Person.example  
27.     val j = parse(s)  
28.     j.extract[Person]  
29.   }  
30. }  

package sample

import org.json4s._
import org.json4s.native.JsonMethods._

object Json4sNativeSample {
  implicit val formats = DefaultFormats

  def jsonSimple() = {
    val s = Company.example
    parse(s)
  }

  def jsonComplex() = {
    val s = Person.example
    parse(s)
  }

  def simple() = {
    val s = Company.example
    val j = parse(s)
    j.extract[Company]
  }

  def complex() = {
    val s = Person.example
    val j = parse(s)
    j.extract[Person]
  }
}

Json4s jackson

view plainprint?

1. package sample  
2.   
3. import org.json4s._  
4. import org.json4s.jackson.JsonMethods._  
5.   
6. object Json4sSample {  
7.   implicit val formats = DefaultFormats  
8.   
9.   def jsonSimple() = {  
10.     val s = Company.example  
11.     parse(s)  
12.   }  
13.   
14.   def jsonComplex() = {  
15.     val s = Person.example  
16.     parse(s)  
17.   }  
18.   
19.   def simple() = {  
20.     val s = Company.example  
21.     val j = parse(s)  
22.     j.extract[Company]  
23.   }  
24.   
25.   def complex() = {  
26.     val s = Person.example  
27.     val j = parse(s)  
28.     j.extract[Person]  
29.   }  
30. }  

package sample

import org.json4s._
import org.json4s.jackson.JsonMethods._

object Json4sSample {
  implicit val formats = DefaultFormats

  def jsonSimple() = {
    val s = Company.example
    parse(s)
  }

  def jsonComplex() = {
    val s = Person.example
    parse(s)
  }

  def simple() = {
    val s = Company.example
    val j = parse(s)
    j.extract[Company]
  }

  def complex() = {
    val s = Person.example
    val j = parse(s)
    j.extract[Person]
  }
}

Play-json

view plainprint?

1. package sample  
2.   
3. import play.api.libs.json._  
4.   
5. object PlaySample {  
6.   implicit val companyReads = Json.reads[Company]  
7.   implicit val addressReads = Json.reads[Address]  
8.   implicit val personReads = Json.reads[Person]  
9.   
10.   def jsonSimple() = {  
11.     val s = Company.example  
12.     Json.parse(s)  
13.   }  
14.   
15.   def jsonComplex() = {  
16.     val s = Person.example  
17.     Json.parse(s)  
18.   }  
19.   
20.   def simple() = {  
21.     val s = Company.example  
22.     val j = Json.parse(s)  
23.     Json.fromJson[Company](j)  
24.   }  
25.   
26.   def complex() = {  
27.     val s = Person.example  
28.     val j = Json.parse(s)  
29.     Json.fromJson[Person](j)  
30.   }  
31. }  

package sample

import play.api.libs.json._

object PlaySample {
  implicit val companyReads = Json.reads[Company]
  implicit val addressReads = Json.reads[Address]
  implicit val personReads = Json.reads[Person]

  def jsonSimple() = {
    val s = Company.example
    Json.parse(s)
  }

  def jsonComplex() = {
    val s = Person.example
    Json.parse(s)
  }

  def simple() = {
    val s = Company.example
    val j = Json.parse(s)
    Json.fromJson[Company](j)
  }

  def complex() = {
    val s = Person.example
    val j = Json.parse(s)
    Json.fromJson[Person](j)
  }
}

Argonaut

view plainprint?

1. package sample  
2.   
3. import scalaz._, Scalaz._  
4. import argonaut._, Argonaut._  
5.   
6. object ArgonautSample {  
7.   implicit def decodeCompanyJson: DecodeJson[Company] =  
8.     casecodec1(Company.apply, Company.unapply)("name")  
9.   
10.   implicit def encodeCompanyJson: EncodeJson[Company] =  
11.     jencode1L((a: Company) => (a.name))("name")  
12.   
13.   implicit def decodeAddressJson: DecodeJson[Address] =  
14.     casecodec2(Address.apply, Address.unapply)("zip", "city")  
15.   
16.   implicit def encodeAddressJson: EncodeJson[Address] =  
17.     jencode2L((a: Address) => (a.zip, a.city))("zip", "city")  
18.   
19.   implicit def decodePersonJson: DecodeJson[Person] =  
20.     casecodec4(Person.apply, Person.unapply)("name", "age", "address", "company")  
21.   
22.   def jsonSimple() = {  
23.     val s = Company.example  
24.     Parse.parse(s)  
25.    }  
26.   
27.   def jsonComplex() = {  
28.     val s = Person.example  
29.     Parse.parse(s)  
30.    }  
31.   
32.   def simple() = {  
33.     val s = Company.example  
34.     Parse.decodeOption[Company](s)  
35.   }  
36.   
37.   def complex() = {  
38.     val s = Person.example  
39.     Parse.decodeOption[Person](s)  
40.   }  
41. }  

package sample

import scalaz._, Scalaz._
import argonaut._, Argonaut._

object ArgonautSample {
  implicit def decodeCompanyJson: DecodeJson[Company] =
    casecodec1(Company.apply, Company.unapply)("name")

  implicit def encodeCompanyJson: EncodeJson[Company] =
    jencode1L((a: Company) => (a.name))("name")

  implicit def decodeAddressJson: DecodeJson[Address] =
    casecodec2(Address.apply, Address.unapply)("zip", "city")

  implicit def encodeAddressJson: EncodeJson[Address] =
    jencode2L((a: Address) => (a.zip, a.city))("zip", "city")

  implicit def decodePersonJson: DecodeJson[Person] =
    casecodec4(Person.apply, Person.unapply)("name", "age", "address", "company")

  def jsonSimple() = {
    val s = Company.example
    Parse.parse(s)
   }

  def jsonComplex() = {
    val s = Person.example
    Parse.parse(s)
   }

  def simple() = {
    val s = Company.example
    Parse.decodeOption[Company](s)
  }

  def complex() = {
    val s = Person.example
    Parse.decodeOption[Person](s)
  }
}

テストデータ

Company

view plainprint?

1. package sample  
2.   
3. case class Company(name: String)  
4.   
5. object Company {  
6.   val example = """  
7. {"name": "Yamada Corp"}  
8. """  
9. }  

package sample

case class Company(name: String)

object Company {
  val example = """
{"name": "Yamada Corp"}
"""
}

Person

view plainprint?

1. package sample  
2.   
3. case class Person(name: String, age: Int, address: Address, company: Option[Company])  
4. case class Address(zip: String, city: String)  
5.   
6. object Person {  
7.   val example = """  
8. {"name": "Yamada Taro", "age": 30, "address": {"zip": "045", "city": "Yokohama", "company": "Yamada Corp"}}  
9. """  
10. }  

package sample

case class Person(name: String, age: Int, address: Address, company: Option[Company])
case class Address(zip: String, city: String)

object Person {
  val example = """
{"name": "Yamada Taro", "age": 30, "address": {"zip": "045", "city": "Yokohama", "company": "Yamada Corp"}}
"""
}

測定結果

性能測定は以下のプログラムで行いました。JSONのパース処理を100万回繰り返した時の累積ミリ秒を表示しています。

view plainprint?

1. package sample  
2.   
3. object All {  
4.   def go(msg: String)(body: => Unit): Unit = {  
5.     System.gc  
6.     val ts = System.currentTimeMillis  
7.     for (i <- 0 until 1000000) yield {  
8.       body  
9.     }  
10.     println(s"$msg:${System.currentTimeMillis - ts}")  
11.   }  
12.   
13.   def main(args: Array[String]) {  
14.     go(s"Json4s native json simple")(Json4sNativeSample.jsonSimple)  
15.     go(s"Json4s jackson json simple")(Json4sSample.jsonSimple)  
16.     go(s"Play json simple")(PlaySample.jsonSimple)  
17.     go(s"Argonaut json simple")(ArgonautSample.jsonSimple)  
18.     go(s"Json4s native json complex")(Json4sNativeSample.jsonComplex)  
19.     go(s"Json4s jackson json complex")(Json4sSample.jsonComplex)  
20.     go(s"Play json complex")(PlaySample.jsonComplex)  
21.     go(s"Argonaut json complex")(ArgonautSample.jsonComplex)  
22.     go(s"Json4s native simple")(Json4sNativeSample.simple)  
23.     go(s"Json4s jackson simple")(Json4sSample.simple)  
24.     go(s"Play simple")(PlaySample.simple)  
25.     go(s"Argonaut simple")(ArgonautSample.simple)  
26.     go(s"Json4s native complex")(Json4sNativeSample.complex)  
27.     go(s"Json4s jackson complex")(Json4sSample.complex)  
28.     go(s"Play complex")(PlaySample.complex)  
29.     go(s"Argonaut complex")(ArgonautSample.complex)  
30.   }  
31. }  

package sample

object All {
  def go(msg: String)(body: => Unit): Unit = {
    System.gc
    val ts = System.currentTimeMillis
    for (i <- 0 until 1000000) yield {
      body
    }
    println(s"$msg: ${System.currentTimeMillis - ts}")
  }

  def main(args: Array[String]) {
    go(s"Json4s native json simple")(Json4sNativeSample.jsonSimple)
    go(s"Json4s jackson json simple")(Json4sSample.jsonSimple)
    go(s"Play json simple")(PlaySample.jsonSimple)
    go(s"Argonaut json simple")(ArgonautSample.jsonSimple)
    go(s"Json4s native json complex")(Json4sNativeSample.jsonComplex)
    go(s"Json4s jackson json complex")(Json4sSample.jsonComplex)
    go(s"Play json complex")(PlaySample.jsonComplex)
    go(s"Argonaut json complex")(ArgonautSample.jsonComplex)
    go(s"Json4s native simple")(Json4sNativeSample.simple)
    go(s"Json4s jackson simple")(Json4sSample.simple)
    go(s"Play simple")(PlaySample.simple)
    go(s"Argonaut simple")(ArgonautSample.simple)
    go(s"Json4s native complex")(Json4sNativeSample.complex)
    go(s"Json4s jackson complex")(Json4sSample.complex)
    go(s"Play complex")(PlaySample.complex)
    go(s"Argonaut complex")(ArgonautSample.complex)
  }
}

性能測定結果は以下になります。

ネストなし/JSON

ライブラリ	性能(ms)
Json4s native	747
Json4s jackson	788
Play-json	640
Argonaut	872

ほとんど同じですがPlay-jsonが速く、Argonautが遅いという結果になっています。

ネストあり/JSON

ライブラリ	性能(ms)
Json4s native	1495
Json4s jackson	1310
Play-json	1535
Argonaut	1724

ネストありのJSONの場合は、Json4s jacksonが速く、Argonautが遅いという結果になりました。

ネストなし/case class

ライブラリ	性能(ms)
Json4s native	2080
Json4s jackson	1601
Play-json	1149
Argonaut	1151

JSONをcase classにマッピングする場合は、単にJSONをパースするだけの性能とは全く異なる性能特性になりました。

Json4s系はいずれもかなり遅くなっています。

Play-jsonとArgonautはほぼ同じで高速です。

ネストあり/case class

ライブラリ	性能(ms)
Json4s native	5162
Json4s jackson	4786
Play-json	4471
Argonaut	3840

ネストありのJSONをcase classにマッピングする場合は、Argonautが高速でした。

Json4sはこのケースでもかなり遅くなっています。

評価

性能測定結果を評価の観点で表にまとめました。

ライブラリ	ネストありJSON性能	case class性能	case class設定
Json4s native	◯	△	◯
Json4s jackson	◎	△	◯
Play-json	◯	◯	◯
Argonaut	△	◎	△

ネストありJSONのパース性能はJson4sが最速ですが、逆にcase classへのマッピングはかなり遅くなっています。このため、Json4s一択という感じではなさそうです。

ArgonautはJSONパースは遅く、case classへのマッピング性能はよいもののマッピング設定が煩雑なので、使い場所が難しい感触です。Scalazと組合せて関数型的に面白いことが色々できそうなので、その点をどう評価するかということだと思います。

一番バランスがよいのがPlay-jsonで、一択するならPlay-jsonがよさそうです。

ただPlay-jsonはPlayのバージョンと連動しているためdependency hellの可能性があるのが難点です。たとえば、最新版のPlay-jsonを使って共通ライブラリを作っても、サービスで運用しているPlayのバージョンが古いと適用できない、といった問題です。

まとめ

全体的にPlay-jsonが最もバランスがよいという結果が得られました。

個別の性能的には「ユースケース1: 複雑な構造を持ったJSON→CSV」はJson4s jackson、「ユースケース2: 複雑な構造を持ったJSON→アプリケーションロジック適用」はArgonautが適しているという結果が得られました。

ただArgonautはcase classの設定が煩雑なので、広範囲に使える共通処理を作る場合はよいのですが、用途ごとの小さな要件には適用しづらい感じです。そういう意味では、Play-jsonがこの用途にも適していそうです。

以上の点から、次のような方針で考えていこうと思います。

• Play-jsonを基本にする
• 必要に応じて「大規模JSON→CSV」の用途向けにJson4s jacksonを併用
• 必要に応じて「大規模JSON→アプリケーションロジック適用」の用途向けにArgonautを併用

Play-jsonを基本ライブラリとして採用するメリットが大きいので、Play-jsonのdependency hell問題は、運用で回避する作戦を取りたいと思います。

諸元

• Mac OS 10.7.5 (2.6 GHz Intel Core i7)
• Java 1.7.0_75
• Scala 2.11.6
• Json4s native 2.3.11
• Json4s jackson 2.3.11
• Play-json 2.3.9
• Argonaut 6.1

case class 2015

case classはScalaプログラミングにおける最重要機能の一つです。

case classはいろいろな便利機能が言語機能としてビルトインされているのに加えて、OOP的にはValue Object、関数型プログラミングとしては代数的データ構造として利用することができます。

case classはそのままシンプルに使っても便利ですが、case classを作る時に基本対応しておくとよさそうな各種機能があるので、その辺りの最新事情を取り込んだ2015年版case classについて考えてみます。

対応機能

2015年版case classでは以下の機能に対応することにします。

• Monoid
• Argonaut
• ScalaCheck

Monoid

「Scala Tips / Monoid - 新規作成」は2012年6月の記事なので、かれこれ3年になりますがMonoidの重要性は変わる所がありません。Monoidを使うためだけにScalazを導入してもお釣りがくるぐらいです。

case classを作る時は常にMonoidを意識しておいて、可能であればMonoid化しておくのがよいでしょう。

MonoidはScalazで実装します。

2012年版の「Scala Tips / Monoid - 新規作成」ではScalaz 6でしたが、今回はScalaz 7でMonoidの定義の仕方も変更されています。

Argonaut

Finagleなどを使ってRESTベースのmicroservicesアーキテクチャを取る場合、case classをJSONで送受信するニーズが大きくなります。

case classをできるだけ簡単にJSON化する方法としてArgonautが有力なので使ってみました。

ScalaCheck

case classがMonoidである場合は、必ず二項演算があるので、この二項演算のテストコードが必要になります。

Scalaプログラミングでは、こういった演算はScalaCheckでプロパティベーステストを行うのがお約束になっています。

プログラム

build.sbt

build.sbtは特に難しい所はありません。必要なライブラリを登録しているだけです。

view plainprint?

1. scalaVersion := "2.11.6"  
2.   
3. val scalazVersion = "7.1.0"  
4.   
5. libraryDependencies ++= Seq(  
6.   "org.scalaz" %% "scalaz-core" % scalazVersion,  
7.   "io.argonaut" %% "argonaut" % "6.1-M4",  
8.   "org.scalatest" %% "scalatest" % "2.2.4" % "test",  
9.   "org.scalacheck" %% "scalacheck" % "1.12.2" % "test",  
10.   "junit" % "junit" % "4.12" % "test"  
11. )  

scalaVersion := "2.11.6"

val scalazVersion = "7.1.0"

libraryDependencies ++= Seq(
  "org.scalaz" %% "scalaz-core" % scalazVersion,
  "io.argonaut" %% "argonaut" % "6.1-M4",
  "org.scalatest" %% "scalatest" % "2.2.4" % "test",
  "org.scalacheck" %% "scalacheck" % "1.12.2" % "test",
  "junit" % "junit" % "4.12" % "test"
)

Average.scala

Monoid化とArgonaut化したcase class Averageは以下になります。

view plainprint?

1. package sample  
2.   
3. import scalaz._, Scalaz._  
4. import argonaut._, Argonaut._  
5.   
6. case class Average(count: Int, total: Int) {  
7.   import Average.Implicits._  
8.   
9.   def value: Float = total / count.toFloat  
10.   
11.   def +(rhs: Average): Average = {  
12.     Average(count + rhs.count, total + rhs.total)  
13.   }  
14.   
15.   def marshall: String = this.asJson.nospaces  
16. }  
17.   
18. object Average {  
19.   import Implicits._  
20.   
21.   val empty = Average(0, 0)  
22.   
23.   def unmarshall(s: String): Validation[String, Average] = s.decodeValidation[Average]  
24.   
25.   object Implicits {  
26.     implicit object AverageMonoid extends Monoid[Average] {  
27.       def append(lhs: Average, rhs: => Average) = lhs + rhs  
28.       def zero = Average.empty  
29.     }  
30.   
31.     implicit def decodeAverageJson: DecodeJson[Average] =  
32.       casecodec2(Average.apply, Average.unapply)("count", "total")  
33.   
34.     implicit def encodeAverageJson: EncodeJson[Average] =  
35.       jencode2L((d: Average) => (d.count, d.total))("count", "total")  
36.   }  
37. }  

package sample

import scalaz._, Scalaz._
import argonaut._, Argonaut._

case class Average(count: Int, total: Int) {
  import Average.Implicits._

  def value: Float = total / count.toFloat

  def +(rhs: Average): Average = {
    Average(count + rhs.count, total + rhs.total)
  }

  def marshall: String = this.asJson.nospaces
}

object Average {
  import Implicits._

  val empty = Average(0, 0)

  def unmarshall(s: String): Validation[String, Average] = s.decodeValidation[Average]

  object Implicits {
    implicit object AverageMonoid extends Monoid[Average] {
      def append(lhs: Average, rhs: => Average) = lhs + rhs
      def zero = Average.empty
    }

    implicit def decodeAverageJson: DecodeJson[Average] =
      casecodec2(Average.apply, Average.unapply)("count", "total")

    implicit def encodeAverageJson: EncodeJson[Average] =
      jencode2L((d: Average) => (d.count, d.total))("count", "total")
  }
}

case classの定義に難しいところはないと思います。

以下ではMonoidとArgonautの定義について説明します。

Monoid

Monoidは、Scalazの型クラスMonoidの型クラスインスタンスを作成して暗黙オブジェクトとして定義します。

view plainprint?

1. implicit object AverageMonoid extends Monoid[Average] {  
2.       def append(lhs: Average, rhs: => Average) = lhs + rhs  
3.       def zero = Average.empty  
4.     }  

implicit object AverageMonoid extends Monoid[Average] {
      def append(lhs: Average, rhs: => Average) = lhs + rhs
      def zero = Average.empty
    }

型クラスMonoidの型クラスインスタンスのappendメソッドとzeroメソッドを実装します。

appendメソッドはMonoid対象の二項演算を定義します。通常、Monoid対象となるcase classで「+」メソッドを提供しているはずなのでこれを呼び出す形になります。

zeroメソッドは空の値を返すようにします。通常、case classの空値はコンパニオン／オブジェクトの変数emptyに定義するのが普通なので、これをzeroメソッドの値として返す形になります。

Argonaut

まずArgonautの基本定義として、ScalaオブジェクトとJSON間の相互変換をする暗黙関数を2つ用意します。

ここも自動化できればよいのですが、マクロを使う形になるはずなので、マクロの仕様がフィックスしていない現状だとやや時期尚早かもしれません。

view plainprint?

1. implicit def decodeAverageJson: DecodeJson[Average] =  
2.       casecodec2(Average.apply, Average.unapply)("count", "total")  
3.   
4.     implicit def encodeAverageJson: EncodeJson[Average] =  
5.       jencode2L((d: Average) => (d.count, d.total))("count", "total")  

implicit def decodeAverageJson: DecodeJson[Average] =
      casecodec2(Average.apply, Average.unapply)("count", "total")

    implicit def encodeAverageJson: EncodeJson[Average] =
      jencode2L((d: Average) => (d.count, d.total))("count", "total")

次に、マーシャル関数(Scala→JSON)とアンマーシャル関数(JSON→Scala)を定義します。

まず、マーシャル関数はcase class Averageのメソッドとして定義しました。

view plainprint?

1. def marshall: String = this.asJson.nospaces  

def marshall: String = this.asJson.nospaces

アンマーシャル関数はAverageのコンパニオンオブジェクトに定義しました。

view plainprint?

1. def unmarshall(s: String): Validation[String, Average] = s.decodeValidation[Average]  

def unmarshall(s: String): Validation[String, Average] = s.decodeValidation[Average]

AverageSpec.scala

case class Averageのテストコードとして、ScalaCheckによるプロパティベーステストを行うAverageSpecを作りました。

view plainprint?

1. package sample  
2.   
3. import org.junit.runner.RunWith  
4. import org.scalatest.junit.JUnitRunner  
5. import org.scalatest._  
6. import org.scalatest.prop.GeneratorDrivenPropertyChecks  
7.   
8. @RunWith(classOf[JUnitRunner])  
9. class AverageSpec extends WordSpec with Matchers with GivenWhenThen with GeneratorDrivenPropertyChecks {  
10.   "Average" should {  
11.     "value" in {  
12.       forAll ("ns") { (ns: List[Int]) =>  
13.         val ns1 = ns.filter(_ >= 0)  
14.         if (ns1.nonEmpty) {  
15.           val x = Average(ns1.length, ns1.sum)  
16.           x.value should be (toAverageValue(ns1))  
17.         }  
18.       }  
19.     }  
20.     "value gen" in {  
21.       forAll ((Gen.nonEmptyListOf(Gen.posNum[Int]), "ns")) { ns =>  
22.         val x = Average(ns.length, ns.sum)  
23.         x.value should be (toAverageValue(ns))  
24.       }  
25.     }  
26.     "+" in {  
27.       forAll ("ns") { (ns: List[Int]) =>  
28.         val ns1 = ns.filter(_ >= 0)  
29.         if (ns1.nonEmpty) {  
30.           val xs = toAverages(ns1)  
31.           val r = xs.foldLeft(Average.empty)((z, x) => z + x)  
32.           r.value should be (toAverageValue(ns1))  
33.         }  
34.       }  
35.     }  
36.     "monoid concatenate" in {  
37.       import scalaz._, Scalaz._  
38.       import Average.Implicits.AverageMonoid  
39.       forAll ("ns") { (ns: List[Int]) =>  
40.         val ns1 = ns.filter(_ >= 0)  
41.         if (ns1.nonEmpty) {  
42.           val xs = toAverages(ns1)  
43.           val r = xs.concatenate  
44.           r.value should be (toAverageValue(ns1))  
45.         }  
46.       }  
47.     }  
48.   
49.     def toAverages(ns: List[Int]): List[Average] = {  
50.       ns.map(Average(1, _))  
51.     }  
52.   
53.     def toAverageValue(ns: List[Int]): Float = {  
54.       ns.sum.toFloat / ns.length  
55.     }  
56.   }  
57. }  

package sample

import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
import org.scalatest._
import org.scalatest.prop.GeneratorDrivenPropertyChecks

@RunWith(classOf[JUnitRunner])
class AverageSpec extends WordSpec with Matchers with GivenWhenThen with GeneratorDrivenPropertyChecks {
  "Average" should {
    "value" in {
      forAll ("ns") { (ns: List[Int]) =>
        val ns1 = ns.filter(_ >= 0)
        if (ns1.nonEmpty) {
          val x = Average(ns1.length, ns1.sum)
          x.value should be (toAverageValue(ns1))
        }
      }
    }
    "value gen" in {
      forAll ((Gen.nonEmptyListOf(Gen.posNum[Int]), "ns")) { ns =>
        val x = Average(ns.length, ns.sum)
        x.value should be (toAverageValue(ns))
      }
    }
    "+" in {
      forAll ("ns") { (ns: List[Int]) =>
        val ns1 = ns.filter(_ >= 0)
        if (ns1.nonEmpty) {
          val xs = toAverages(ns1)
          val r = xs.foldLeft(Average.empty)((z, x) => z + x)
          r.value should be (toAverageValue(ns1))
        }
      }
    }
    "monoid concatenate" in {
      import scalaz._, Scalaz._
      import Average.Implicits.AverageMonoid
      forAll ("ns") { (ns: List[Int]) =>
        val ns1 = ns.filter(_ >= 0)
        if (ns1.nonEmpty) {
          val xs = toAverages(ns1)
          val r = xs.concatenate
          r.value should be (toAverageValue(ns1))
        }
      }
    }

    def toAverages(ns: List[Int]): List[Average] = {
      ns.map(Average(1, _))
    }

    def toAverageValue(ns: List[Int]): Float = {
      ns.sum.toFloat / ns.length
    }
  }
}

ScalaCheckのプロパティベーステストを行う方法はいくつかありますが、ここではScalaTestのGeneratorDrivenPropertyChecksを使ってみました。

GeneratorDrivenPropertyChecksを使うと「forAll」を使って、指定された型の値をワンセット自動生成してくれるので、この値を用いてテストを行うことができます。

forAllの内部定義として個々のテストを書いていきますが、これは通常のテストコードと同様です。

view plainprint?

1. "value" in {  
2.       forAll ("ns") { (ns: List[Int]) =>  
3.         val ns1 = ns.filter(_ >= 0)  
4.         if (ns1.nonEmpty) {  
5.           val x = Average(ns1.length, ns1.sum)  
6.           x.value should be (toAverageValue(ns1))  
7.         }  
8.       }  
9.     }  

"value" in {
      forAll ("ns") { (ns: List[Int]) =>
        val ns1 = ns.filter(_ >= 0)
        if (ns1.nonEmpty) {
          val x = Average(ns1.length, ns1.sum)
          x.value should be (toAverageValue(ns1))
        }
      }
    }

一つポイントとなるのは、テスト用データの自動生成は指定された型(ここでは List[Int])の任意の値を取る可能性があるので、これをテストコード側で排除する必要がある点です。

この問題への対処方法として、テスト用データ生成器(org.scalacheck.Gen)で値域を指定する方法があります。

org.scalacheck.Genを使って値域を指定するテスト"value gen"は以下になります。org.scalacheck.Genを使うとクロージャの引数の型(List[Int])も省略できます。

view plainprint?

1. "value gen" in {  
2.       forAll ((Gen.nonEmptyListOf(Gen.posNum[Int]), "ns")) { ns =>  
3.         val x = Average(ns.length, ns.sum)  
4.         x.value should be (toAverageValue(ns))  
5.       }  
6.     }  

"value gen" in {
      forAll ((Gen.nonEmptyListOf(Gen.posNum[Int]), "ns")) { ns =>
        val x = Average(ns.length, ns.sum)
        x.value should be (toAverageValue(ns))
      }
    }

いずれの方法を取るにしても、テストプログラムを書く時に、テストデータを準備する必要はないのは大変便利です。

また、テストプログラムが、テストをする手続きというより、より仕様定義に近いものになるのもよい感触です。

実行

Sbtのtestでテストプログラムを実行することができます。

$ sbt test
...略...
[info] AverageSpec:
[info] Average
[info] - should value
[info] - should value gen
[info] - should +
[info] - should monoid concatenate
[info] ScalaTest
[info] 36mRun completed in 1 second, 870 milliseconds.0m
[info] 36mTotal number of tests run: 40m
[info] 36mSuites: completed 1, aborted 00m
[info] 36mTests: succeeded 4, failed 0, canceled 0, ignored 0, pending 00m
[info] 32mAll tests passed.0m
[info] Passed: Total 4, Failed 0, Errors 0, Passed 4
[success] Total time: 5 s, completed 2015/05/15 16:31:57

まとめ

case classを作る時に意識しておきたい基本形を考えてみました。

プログラミング時には、この基本形を念頭に置いて、不要な機能は削除、必要な機能を追加という形でcase classを組み立てていくことをイメージしています。

なお、Monoidに対するテストに関してはScalazのSpecLiteを使うともっと強力なテストができますが、この話題は別途取り上げたいと思います。

諸元

• Scala 2.11.6
• Scalaz 7.1.0
• Argonaut 6.1-M4
• ScalaTest 2.2.4
• ScalaCheck 1.12.2

[docker] DockerでSpark SQL

Spark SQL 1.3から以下の2つの機能が導入されています。

• DataSourceとしてJDBCが使えるようになった
• DataFrame

この2つの機能追加によってSpark SQLを汎用のバッチ処理基盤にできるのではないかというインスピレーションが湧きました。

この実現目的でSparkバッチをスタンドアロンで実行するためのDockerイメージspark-sql-scala-dockerを作ってみた、というのが今回のお話です。

Spark SQL

Spark SQLは、Sparkの分散計算処理をSQLで記述できるようにしたものです。SQLとSpark本来のmonadicなAPI(e.g. filter, map, flatMap)を併用して計算処理を記述することができます。

このプログラミングモデルは非常に強力で、大枠の絞り込みはSQLで行っておいて、アプリケーションに特化した検索ロジックをScalaで記述したUDF(User Defined Function)で補完するといった処理を、プログラミング言語的に簡潔に記述することができます。

Spark SQLの基本機能に加えて1.3から以下の機能も使えるようになりました。

DataSourceのJDBC対応

DataSourceとしてJDBCが使えるようになったことで、RedShift上にためた分析データなどから直接データを取得できるようになりました。MySQLやPostreSQLなどのデータを一旦S3に変換するといった準備タスクが不要になったので、ジョブ作成の手間が大きく低減すると思います。

小さな機能追加ですが、実運用上のインパクトは大きいのではないかと思います。

DataFrame

大きな機能追加としてはDataFrameが導入されました。

DataFrameは表形式の大規模データを抽象化したAPIで、元々はR/Pythonで実績のある機能のようです。

DataFrameは分析専用のAPIではなく、表形式データ操作の汎用APIとして使用できるのではないかと期待しています。計算結果を外部出力する際の汎用機能としても期待できます。

もちろんR/Pythonなどのデータ分析処理系との連携も期待できそうです。

Spark SQLの用途

Spark SQLの基本機能と上記の2つの機能追加によって、Sparkバッチを大規模(データ量/計算量)向けデータ処理基盤としてだけではなく、汎用のバッチ実行基盤として使えるようになるのではないかとインスピレーションが湧いたわけです。

データ集計用のバッチをSparkバッチとして作成して、データ量、計算量に応じてスタンドアロンジョブとSparkクラスタ上でのジョブのいずれかでジョブ実行するというユースケースです。

そのベースとして、Sparkクラスタを用いないスタンドアロンジョブとして実行するためのDockerイメージを作ってみました。

spark-sql-scala-docker

spark-sql-scala-dockerはSparkアプリケーションをスタンドアロンで実行するためのDockerイメージです。

GitHubにソースコードがありますので、詳細はこちらを参照して下さい。

• https://github.com/asami/spark-sql-scala-docker

以下では、実装上のポイントと使い方について説明します。

Dockerfile

spark-sql-scala-dockerのDockerfileは以下になります。

view plainprint?

1. FROM sequenceiq/spark:1.3.0  
2.   
3. RUN mkdir -p /opt/spark/lib  
4. RUN cd /opt/spark/lib && curl -L 'http://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-java-5.1.30.tar.gz' -o - | tar -xz --strip-components=1 mysql-connector-java-5.1.30/mysql-connector-java-5.1.30-bin.jar  
5. RUN curl -L 'http://jdbc.postgresql.org/download/postgresql-9.2-1002.jdbc4.jar' -o /opt/spark/lib/postgresql-9.2-1002.jdbc4.jar  
6.   
7. ENV SPARK_CLASSPATH /opt/spark/lib/mysql-connector-java-5.1.30-bin.jar:/opt/spark/lib/postgresql-9.2-1002.jdbc4.jar  
8.   
9. RUN rpm -ivh http://ftp-srv2.kddilabs.jp/Linux/distributions/fedora/epel/6/x86_64/epel-release-6-8.noarch.rpm  
10.   
11. RUN yum -y install redis --enablerepo=epel  
12.   
13. COPY spark-defaults.conf /opt/spark-defaults.conf  
14.   
15. COPY entrypoint.sh /opt/entrypoint.sh  
16.   
17. ENV COMMAND_JAR_DIR /opt/command.d  
18.   
19. ENV COMMAND_JAR_NAME command.jar  
20.   
21. VOLUME [$COMMAND_JAR_DIR"]  
22.   
23. ENTRYPOINT ["/opt/entrypoint.sh"]  

FROM sequenceiq/spark:1.3.0

RUN mkdir -p /opt/spark/lib
RUN cd /opt/spark/lib && curl -L 'http://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-java-5.1.30.tar.gz' -o - | tar -xz --strip-components=1 mysql-connector-java-5.1.30/mysql-connector-java-5.1.30-bin.jar
RUN curl -L 'http://jdbc.postgresql.org/download/postgresql-9.2-1002.jdbc4.jar' -o /opt/spark/lib/postgresql-9.2-1002.jdbc4.jar

ENV SPARK_CLASSPATH /opt/spark/lib/mysql-connector-java-5.1.30-bin.jar:/opt/spark/lib/postgresql-9.2-1002.jdbc4.jar

RUN rpm -ivh http://ftp-srv2.kddilabs.jp/Linux/distributions/fedora/epel/6/x86_64/epel-release-6-8.noarch.rpm

RUN yum -y install redis --enablerepo=epel

COPY spark-defaults.conf /opt/spark-defaults.conf

COPY entrypoint.sh /opt/entrypoint.sh

ENV COMMAND_JAR_DIR /opt/command.d

ENV COMMAND_JAR_NAME command.jar

VOLUME [$COMMAND_JAR_DIR"]

ENTRYPOINT ["/opt/entrypoint.sh"]

Dockerイメージsequenceiq/spark:1.3.0をベースにしていて以下の調整だけ行っています。

• MySQLとPostgreSQLのJDBCドライバのインストール
• Sparkアプリケーションの登録処理

entrypoint.sh

spark-sql-scala-dockerのentrypoint.shは以下になります。

view plainprint?

1. #! /bin/bash  
2.   
3. # WAIT_CONTAINER_TIMER  
4. # WAIT_CONTAINER_FILE  
5. # WAIT_CONTAINER_KEY  
6.   
7. # set -x  
8.   
9. set -e  
10.   
11. echo MySQL host: ${MYSQL_SERVER_HOST:=$MYSQL_PORT_3306_TCP_ADDR}  
12. echo MySQL port: ${MYSQL_SERVER_PORT:=$MYSQL_PORT_3306_TCP_PORT}  
13. echo PostgreSQL host: ${POSTGRESQL_SERVER_HOST:=$POSTGRESQL_PORT_5432_TCP_ADDR}  
14. echo PostgreSQL port: ${POSTGRESQL_SERVER_PORT:=$POSTGRESQL_PORT_5432_TCP_PORT}  
15. echo Redis host: ${REDIS_SERVER_HOST:=$REDIS_PORT_6379_TCP_ADDR}  
16. echo Redis port: ${REDIS_SERVER_PORT:=$REDIS_PORT_6379_TCP_PORT}  
17. export MYSQL_SERVER_HOST  
18. export MYSQL_SERVER_PORT  
19. export POSTGRESQL_SERVER_HOST  
20. export POSTGRESQL_SERVER_PORT  
21. export REDIS_SERVER_HOST  
22. export REDIS_SERVER_PORT  
23.   
24. function wait_container {  
25.     if [ -n "$REDIS_SERVER_HOST" ]; then  
26.  wait_container_redis  
27.     elif [ -n "$WAIT_CONTAINER_FILE" ]; then  
28.  wait_container_file  
29.     fi  
30. }  
31.   
32. function wait_container_redis {  
33.     result=1  
34.     for i in $(seq 1 ${WAIT_CONTAINER_TIMER:-100})  
35.     do  
36.  sleep 1s  
37.  result=0  
38.  if [ $(redis-cli -h$REDIS_SERVER_HOST -p $REDIS_SERVER_PORT GET$WAIT_CONTAINER_KEY)'' = "up" ]; then  
39.      break  
40.  fi  
41.  echo spark-sql-scala-docker wait: $REDIS_SERVER_HOST  
42.  result=1  
43.     done  
44.     if [ $result = 1 ]; then  
45.  exit 1  
46.     fi  
47. }  
48.   
49. function wait_container_file {  
50.     result=1  
51.     for i in $(seq 1 ${WAIT_CONTAINER_TIMER:-100})  
52.     do  
53.  sleep 1s  
54.  result=0  
55.  if [ -e $WAIT_CONTAINER_FILE ]; then  
56.      break  
57.  fi  
58.  echo spark-sql-scala-docker wait: $WAIT_CONTAINER_FILE  
59.  result=1  
60.     done  
61.     if [ $result = 1 ]; then  
62.  exit 1  
63.     fi  
64. }  
65.   
66. COMMAND_JAR=$COMMAND_JAR_DIR/$COMMAND_JAR_NAME  
67.   
68. wait_container  
69.   
70. sed -i "s!hdfs://.*:9000!file:\/\/\/tmp!g" /usr/local/hadoop/etc/hadoop/core-site.xml  
71.   
72. spark-submit --properties-file /opt/spark-defaults.conf $COMMAND_JAR  

#! /bin/bash

# WAIT_CONTAINER_TIMER
# WAIT_CONTAINER_FILE
# WAIT_CONTAINER_KEY

# set -x

set -e

echo MySQL host: ${MYSQL_SERVER_HOST:=$MYSQL_PORT_3306_TCP_ADDR}
echo MySQL port: ${MYSQL_SERVER_PORT:=$MYSQL_PORT_3306_TCP_PORT}
echo PostgreSQL host: ${POSTGRESQL_SERVER_HOST:=$POSTGRESQL_PORT_5432_TCP_ADDR}
echo PostgreSQL port: ${POSTGRESQL_SERVER_PORT:=$POSTGRESQL_PORT_5432_TCP_PORT}
echo Redis host: ${REDIS_SERVER_HOST:=$REDIS_PORT_6379_TCP_ADDR}
echo Redis port: ${REDIS_SERVER_PORT:=$REDIS_PORT_6379_TCP_PORT}
export MYSQL_SERVER_HOST
export MYSQL_SERVER_PORT
export POSTGRESQL_SERVER_HOST
export POSTGRESQL_SERVER_PORT
export REDIS_SERVER_HOST
export REDIS_SERVER_PORT

function wait_container {
    if [ -n "$REDIS_SERVER_HOST" ]; then
 wait_container_redis
    elif [ -n "$WAIT_CONTAINER_FILE" ]; then
 wait_container_file
    fi
}

function wait_container_redis {
    result=1
    for i in $(seq 1 ${WAIT_CONTAINER_TIMER:-100})
    do
 sleep 1s
 result=0
 if [ $(redis-cli -h $REDIS_SERVER_HOST -p $REDIS_SERVER_PORT GET $WAIT_CONTAINER_KEY)'' = "up" ]; then
     break
 fi
 echo spark-sql-scala-docker wait: $REDIS_SERVER_HOST
 result=1
    done
    if [ $result = 1 ]; then
 exit 1
    fi
}

function wait_container_file {
    result=1
    for i in $(seq 1 ${WAIT_CONTAINER_TIMER:-100})
    do
 sleep 1s
 result=0
 if [ -e $WAIT_CONTAINER_FILE ]; then
     break
 fi
 echo spark-sql-scala-docker wait: $WAIT_CONTAINER_FILE
 result=1
    done
    if [ $result = 1 ]; then
 exit 1
    fi
}

COMMAND_JAR=$COMMAND_JAR_DIR/$COMMAND_JAR_NAME

wait_container

sed -i "s!hdfs://.*:9000!file:\/\/\/tmp!g" /usr/local/hadoop/etc/hadoop/core-site.xml

spark-submit --properties-file /opt/spark-defaults.conf $COMMAND_JAR

基本的にはspark-submitでSparkアプリケーションのジョブをサブミットしているだけですが、以下の2つの調整を行っています。

• Redisを使って他のコンテナの待ち合わせ
• 中間データのローディング先をHDFSではなくローカルファイルに変更する

コンテナの待ち合わせ

Sparkアプリケーションを動作させる前の準備を他のコンテナで進める場合は、コンテナの待ち合わせが必要になります。この待ち合わせをmysql-java-embulk-dockerと同様にRedisを用いて実現しています。

典型的な使用例は、Sparkアプリケーションのテスト実行時でのテストDBの準備です。この実例は後ほどサンプルで説明します。

中間データのローディング先

core-site.xmlの変更処理です。

view plainprint?

1. sed -i "s!hdfs://.*:9000!file:\/\/\/tmp!g" /usr/local/hadoop/etc/hadoop/core-site.xml  

sed -i "s!hdfs://.*:9000!file:\/\/\/tmp!g" /usr/local/hadoop/etc/hadoop/core-site.xml

ここの設定を変更しないとDocker環境内でスタンドアロンでは動かなかったので設定変更しています。

設定変更の方法としてはHDFSを動くようにするという方式もあるのですが、スタンドアプリケーションなのでここではローカルのファイルを使う方式で対応しています。

Docker Hub

mysql-java-embulk-dockerと同様にspark-sql-scala-dockerもDocker Hubの自動ビルドの設定を行っているので、以下の場所にDockerイメージが自動ビルドされます。

• https://registry.hub.docker.com/u/asami/spark-sql-scala-docker/

このイメージは「asami/spark-sql-scala-docker」という名前で利用することができます。

サンプル

Dockerイメージ「asami/spark-sql-scala-docker」を利用してテストデータの投入を行うサンプルを作ってみます。

手元の環境上でテスト目的で動作させるためmysql-java-embulk-dockerを併用してテストデータの投入を行っています。

サンプルのコードはGitHubのspark-sql-scala-dockerのsampleディレクトリにあるので、詳細はこちらを参照して下さい。

• https://github.com/asami/spark-sql-scala-docker/tree/master/sample

SimpleApp.scala

サンプルのSparkバッチであるSimpleAppのプログラムは以下になります。

view plainprint?

1. import org.apache.spark.SparkContext  
2. import org.apache.spark.SparkContext._  
3. import org.apache.spark.SparkConf  
4. import org.apache.spark.sql.{SQLContext, DataFrame}  
5.   
6. object SimpleApp extends App {  
7.   val batting = SparkSqlUtils.createMysqlDataFrame("Simple Application", "batting")  
8.   val count = batting.count()  
9.   println(s"count = ${batting.count()}")  
10. }  
11.   
12. object SparkSqlUtils {  
13.   def createSqlContext(name: String): SQLContext = {  
14.     val conf = new SparkConf().setAppName(name)  
15.     val sc = new SparkContext(conf)  
16.     new SQLContext(sc)  
17.   }  
18.   
19.   def createMysqlDataFrame(name: String, table: String): DataFrame = {  
20.     val sqlc = createSqlContext(name)  
21.     createMysqlDataFrame(sqlc, table)  
22.   }  
23.   
24.   def createMysqlDataFrame(sqlc: SQLContext, table: String): DataFrame = {  
25.     val host = System.getenv("MYSQL_SERVER_HOST")  
26.     val port = System.getenv("MYSQL_SERVER_PORT")  
27.     val user = System.getenv("MYSQL_SERVER_USER")  
28.     val password = System.getenv("MYSQL_SERVER_PASSWORD")  
29.     sqlc.load("jdbc", Map(  
30.       "url" -> s"jdbc:mysql://$host:$port/baseball?user=$user&password=$password",  
31.       "dbtable" -> table  
32.     ))  
33.   }  
34. }  

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.sql.{SQLContext, DataFrame}

object SimpleApp extends App {
  val batting = SparkSqlUtils.createMysqlDataFrame("Simple Application", "batting")
  val count = batting.count()
  println(s"count = ${batting.count()}")
}

object SparkSqlUtils {
  def createSqlContext(name: String): SQLContext = {
    val conf = new SparkConf().setAppName(name)
    val sc = new SparkContext(conf)
    new SQLContext(sc)
  }

  def createMysqlDataFrame(name: String, table: String): DataFrame = {
    val sqlc = createSqlContext(name)
    createMysqlDataFrame(sqlc, table)
  }

  def createMysqlDataFrame(sqlc: SQLContext, table: String): DataFrame = {
    val host = System.getenv("MYSQL_SERVER_HOST")
    val port = System.getenv("MYSQL_SERVER_PORT")
    val user = System.getenv("MYSQL_SERVER_USER")
    val password = System.getenv("MYSQL_SERVER_PASSWORD")
    sqlc.load("jdbc", Map(
      "url" -> s"jdbc:mysql://$host:$port/baseball?user=$user&password=$password",
      "dbtable" -> table
    ))
  }
}

SparkSqlUtilsにDataFrame取得処理をまとめています。ここは汎用ライブラリ化できるところです。

この処理を除いた以下の処理がSparkバッチの本体です。

view plainprint?

1. val batting = SparkSqlUtils.createMysqlDataFrame("Simple Application", "batting")  
2.   val count = batting.count()  
3.   println(s"count = ${batting.count()}")  

val batting = SparkSqlUtils.createMysqlDataFrame("Simple Application", "batting")
  val count = batting.count()
  println(s"count = ${batting.count()}")

アプリケーションロジック

指定したテーブル"batting"に対応したDataFrameを取得し、countメソッドでレコード総数を取得し、その結果をコンソールに出力しています。とても簡単ですね。

この部分を以下の機能を用いて記述することで高度なバッチ処理を簡単に記述できます。

• DataFrameによる表データ操作
• DataFrameから変換したRDDを用いてSpark計算処理

前述したように「大枠の絞り込みはSQLで行っておいて、アプリケーションに特化した検索ロジックをScalaで記述したUDF(User Defined Function)で補完するといった処理を、プログラミング言語的に簡潔に記述することができます。」

移入・移出

テーブル"batting"をDataFrameとしてローディングしているのは、前述のSpark 1.3の機能追加「DataSourceとしてJDBCが使えるようになった」によるものです。

また、ここでは外部出力をコンソール出力にしていますが、RDDのsaveAsTextFileメソッドやDataFrameを用いることで、S3やデータベースなどに集計結果を簡単に出力することができます。

データベースなどへの外部出力が簡単に行えるのもSpark 1.3の機能追加「DataFrame」の効果です。

ここからも分かるように、Spark SQL 1.3で導入された「DataSourceとしてJDBCが使えるようになった」と「DataFrame」により、Sparkバッチ処理の難題であったデータの移入／移出処理が極めて簡単に記述できるようになったわけです。

SBTの設定

SBTによるScalaプログラムのビルドの設定は以下になります。

Spark本体とSpark SQLを依存ライブラリとして設定している、ごくオーソドックスな設定です。

Sparkバッチ用にすべての依存ライブラリをまとめたJARファイルを作る必要があるので、sbt-assemblyの設定を行ってます。

ポイントとしては、Sparkバッチの実行環境にScalaの基本ライブラリとSpark本体/Spark SQLのライブラリが用意されているので、sbt-assemblyでまとめるJARファイルから排除する設定を行っています。

• Spark本体とSpark SQLの依存ライブラリの設定を"provided"にしてリンク対象から外す
• sbt-assemblyの設定で"includeScala = false"としてScala基本ライブラリをリンク対象から外す

これらの設定はなくても動作しますが、JARファイルが巨大になってしまいます。

view plainprint?

1. name := "simple"  
2.   
3. version := "1.0"  
4.   
5. scalaVersion := "2.10.4"  
6.   
7. libraryDependencies += "org.apache.spark" %% "spark-core" % "1.3.1" % "provided"  
8.   
9. libraryDependencies += "org.apache.spark" %% "spark-sql" % "1.3.1" % "provided"  
10.   
11. assemblyOption in assembly := (assemblyOption in assembly).value.copy(includeScala = false)  

name := "simple"

version := "1.0"

scalaVersion := "2.10.4"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.3.1" % "provided"

libraryDependencies += "org.apache.spark" %% "spark-sql" % "1.3.1" % "provided"

assemblyOption in assembly := (assemblyOption in assembly).value.copy(includeScala = false)

sbt-assemblyプラグインが必要なのでproject/assembly.sbtに以下の設定をしておきます。

view plainprint?

1. addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.13.0")  

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.13.0")

docker-compose.yml

サンプルプログラムのdocker-compose.ymlは以下になります。

view plainprint?

1. spark:  
2.   image: asami/spark-sql-scala-docker  
3.   links:  
4.     - mysql  
5.     - redis  
6.   volumes:  
7.     - target/scala-2.10:/opt/command.d  
8.   environment:  
9.     COMMAND_JAR_NAME: simple-assembly-1.0.jar  
10.     WAIT_CONTAINER_KEY: mysql-java-embulk-docker  
11.     MYSQL_SERVER_USER: baseball  
12.     MYSQL_SERVER_PASSWORD: baseball  
13. mysql:  
14.   image: asami/mysql-java-embulk-docker  
15.   links:  
16.     - redis  
17.   ports:  
18.     - ":3306"  
19.   volumes:  
20.     - setup.d:/opt/setup.d  
21.   environment:  
22.     MYSQL_USER: baseball  
23.     MYSQL_PASSWORD: baseball  
24.     MYSQL_ROOT_PASSWORD: baseball  
25.     MYSQL_DATABASE: baseball  
26. redis:  
27.   image: redis  
28.   ports:  
29.     - ":6379"  

spark:
  image: asami/spark-sql-scala-docker
  links:
    - mysql
    - redis
  volumes:
    - target/scala-2.10:/opt/command.d
  environment:
    COMMAND_JAR_NAME: simple-assembly-1.0.jar
    WAIT_CONTAINER_KEY: mysql-java-embulk-docker
    MYSQL_SERVER_USER: baseball
    MYSQL_SERVER_PASSWORD: baseball
mysql:
  image: asami/mysql-java-embulk-docker
  links:
    - redis
  ports:
    - ":3306"
  volumes:
    - setup.d:/opt/setup.d
  environment:
    MYSQL_USER: baseball
    MYSQL_PASSWORD: baseball
    MYSQL_ROOT_PASSWORD: baseball
    MYSQL_DATABASE: baseball
redis:
  image: redis
  ports:
    - ":6379"

自前ではDockerイメージを作らず、以下の3つの汎用Dockerイメージを再利用しています。

• asami/spark-sql-scala-docker
• asami/mysql-java-embulk-docker
• redis

asami/spark-sql-scala-docker

ボリュームと環境変数の記述で、targetscala-2.10simple-assembly-1.0.jarがSparkバッチプログラムとして認識されるようにしています。

simple-assembly-1.0.jarはsbt-assemblyで作成した「全部入り(SparkとScala以外)」のJARファイルです。

それ以外は、mysql-java-embulk-dockerと同期をとるためのおまじないです。

asami/mysql-java-embulk-docker

前回の記事「Docker Composeでデータ投入」と同じ設定です。テスト用のMySQLデータベースにテストデータを投入しています。

Batting.csvは以下のサイトからデータを取得しました。

• http://seanlahman.com/baseball-archive/statistics/

redis

asami/mysql-java-embulk-dockerによるテストデータ投入の待ち合わせにredisを用いています。

ビルド

Sparkバッチのビルドはsbtで行います。

$ sbt assembly

テスト環境はdocker-composeのbuildコマンドでビルドします。

$ docker-compose build

実行

docker-composeのupコマンドで実行します。

$ docker-compose up

動作過程がコンソールに出力されますが、最後の方で以下のような出力があります。

spark_1 | count = 99846

無事Sparkバッチでデータ集計ができました。

まとめ

Spark SQLを汎用のバッチ処理基盤として運用する目的でSparkバッチをスタンドアロンで実行するためのDockerイメージspark-sql-scala-dockerを作ってみましたが、無事動作しました。

このことによってspark-sql-scala-dockerとmysql-java-embulk-dockerを使って手元で簡単にSparkバッチをテストできるようになりました。

汎用Dockerイメージをdocker-composeで組み合わせるだけなので運用的にも大変、楽だと思います。

今回は試していませんが、spark-sql-scala-dockerを使ってSparkバッチをECS(EC2 Container Service)などのDocker環境上でスタンドアロンバッチとして実行するという運用も可能ではないかと考えています。

もちろん、SparkバッチのJARファイルをspark-submitコマンドによるジョブ投入により直接Sparkクラスタ上で実行することでSpark本来の大規模(データ量/計算量)処理を行うことができます。

いずれの場合も、基本的に開発するのは、Scalaによる通常のSparkバッチプログラムだけです。テストやDocker環境上でのスタンドアロンバッチのいずれも汎用Dockerイメージを活用することで、簡単な設定のみで運用することができそうです。

今回の作業で上記の3つのユースケースを同時に満たせることの目処が立ちました。この成果をベースにSpark SQLを汎用のバッチ処理基盤として利用するためのノウハウの積み上げをしてきたいと思います。

諸元

• Mac OS 10.7.5
• docker 1.6
• docker-compose 1.2.0
• Spark SQL 1.3

[docker] Docker Composeでデータ投入

「Docker ComposeでMySQLを使う」ではDocker Composeを使ってJobSchedulerからMySQLをそれぞれ別のDockerコンテナ上で動作させ連携して使用しました。

テストなどでMySQL公式Dockerコンテナを使う際の問題点として、データ投入があります。事前に用意したデータをMySQLに投入後に、テスト対象のアプリケーションが起動されると理想的なのですが、MySQL公式Dockerイメージではデータ投入する機能は提供されていません。

また、MySQL公式DockerイメージではMySQLの起動完了を待ち合わせる機能を持っていないことも問題です。

これらの問題に対応するためMySQL公式Dockerイメージを元に、マイDockerイメージであるmysql-java-embulk-dockerを作ってみました。

mysql-java-embulk-docker

mysql-java-embulk-dockerはdocker-composeを使ってアプリケーションDockerイメージのテストを行う際に、事前にデータ投入したMySQLデータベースを提供するためのDockerイメージです。

GitHubにソースコードがありますので、詳細はこちらを参照して下さい。

• https://github.com/asami/mysql-java-embulk-docker

以下では、実装上のポイントと使い方について説明します。

Dockerfile

mysql-java-embulk-dockerのDockerfileは以下になります。

view plainprint?

1. FROM mysql:5.6  
2. MAINTAINER asami  
3.   
4. RUN apt-get update && apt-get -y install wget curl  
5.   
6. # Install JDK 1.7  
7. RUN cd /opt; wget --no-cookies --no-check-certificate --header "Cookie: oraclelicense=accept-securebackup-cookie" "http://download.oracle.com/otn-pub/java/jdk/7u51-b13/jdk-7u51-linux-x64.tar.gz" -O /opt/jdk-7-linux-x64.tar.gz  
8.   
9. # Install in /usr/java/jdk1.7.0_51   
10. RUN mkdir /usr/java && (cd /usr/java; tar xzf /opt/jdk-7-linux-x64.tar.gz)  
11. RUN rm /opt/jdk-7-linux-x64.tar.gz  
12. RUN update-alternatives --install /usr/bin/java java /usr/java/jdk1.7.0_51/jre/bin/java 20000; update-alternatives --install /usr/bin/jar jar /usr/java/jdk1.7.0_51/bin/jar 20000; update-alternatives --install /usr/bin/javac javac /usr/java/jdk1.7.0_51/bin/javac 20000; update-alternatives --install /usr/bin/javaws javaws /usr/java/jdk1.7.0_51/jre/bin/javaws 20000; update-alternatives --set java /usr/java/jdk1.7.0_51/jre/bin/java; update-alternatives --set javaws /usr/java/jdk1.7.0_51/jre/bin/javaws; update-alternatives --set javac /usr/java/jdk1.7.0_51/bin/javac; update-alternatives --set jar /usr/java/jdk1.7.0_51/bin/jar;  
13.   
14. RUN curl --create-dirs -o /opt/embulk -L "http://dl.embulk.org/embulk-latest.jar" && chmod +x /opt/embulk  
15.   
16. RUN /opt/embulk gem install embulk-output-mysql  
17.   
18. RUN apt-get -y install redis-server  
19.   
20. COPY charset.cnf /etc/mysql/conf.d/charset.cnf  
21. COPY entrypoint.sh /opt/entrypoint.sh  
22. RUN chmod +x /opt/entrypoint.sh  
23.   
24. VOLUME ["/var/lib/mysql", "/etc/mysql/conf.d", "/opt/setup.d"]  
25.   
26. ENTRYPOINT ["/opt/entrypoint.sh"]  
27.   
28. CMD ["mysqld"]  

FROM mysql:5.6
MAINTAINER asami

RUN apt-get update && apt-get -y install wget curl

# Install JDK 1.7
RUN cd /opt; wget --no-cookies --no-check-certificate --header "Cookie: oraclelicense=accept-securebackup-cookie" "http://download.oracle.com/otn-pub/java/jdk/7u51-b13/jdk-7u51-linux-x64.tar.gz" -O /opt/jdk-7-linux-x64.tar.gz

# Install in /usr/java/jdk1.7.0_51 
RUN mkdir /usr/java && (cd /usr/java; tar xzf /opt/jdk-7-linux-x64.tar.gz)
RUN rm /opt/jdk-7-linux-x64.tar.gz
RUN update-alternatives --install /usr/bin/java java /usr/java/jdk1.7.0_51/jre/bin/java 20000; update-alternatives --install /usr/bin/jar jar /usr/java/jdk1.7.0_51/bin/jar 20000; update-alternatives --install /usr/bin/javac javac /usr/java/jdk1.7.0_51/bin/javac 20000; update-alternatives --install /usr/bin/javaws javaws /usr/java/jdk1.7.0_51/jre/bin/javaws 20000; update-alternatives --set java /usr/java/jdk1.7.0_51/jre/bin/java; update-alternatives --set javaws /usr/java/jdk1.7.0_51/jre/bin/javaws; update-alternatives --set javac /usr/java/jdk1.7.0_51/bin/javac; update-alternatives --set jar /usr/java/jdk1.7.0_51/bin/jar;

RUN curl --create-dirs -o /opt/embulk -L "http://dl.embulk.org/embulk-latest.jar" && chmod +x /opt/embulk

RUN /opt/embulk gem install embulk-output-mysql

RUN apt-get -y install redis-server

COPY charset.cnf /etc/mysql/conf.d/charset.cnf
COPY entrypoint.sh /opt/entrypoint.sh
RUN chmod +x /opt/entrypoint.sh

VOLUME ["/var/lib/mysql", "/etc/mysql/conf.d", "/opt/setup.d"]

ENTRYPOINT ["/opt/entrypoint.sh"]

CMD ["mysqld"]

Embulk

Embulkはビッグデータスケールのデータローダです。

• http://www.embulk.org/

データ投入にEmbulkを利用できると大変便利なので組み込んでみました。

EmbulkはJava VM上で動作するので、Embulk用にJDKをインストールしています。

また、MySQLにデータ投入するので、Embulkにembulk-output-mysqlプラグインを追加しています。

Redis

Docker Compose上でmysql-java-embulk-dockerコンテナの起動時にデータ投入をする際に問題点として、mysql-java-embulk-dockerコンテナの起動とアプリケーションコンテナの起動の同期が行われないというものがあります。

mysql-java-embulk-dockerコンテナの起動とアプリケーションコンテナの起動が同時に行われてしまうために、mysql-java-embulk-dockerコンテナの起動時に行われるデータ投入が完了する前に、アプリケーションコンテナが動き出してしまい、想定したデータがない状態なので誤動作する、という問題です。

現段階ではDocker Composeにはこの問題を解決するための機能は提供されていないようなので、Redisを使って対応することにしました。

この目的でRedisをインストールしています。

本来はRedisのクライアントのみがインストールできればよいのですが、簡単にできるよい方法がみつからなかったのでRedisをまるごとインストールしています。

charset.cnf

MySQLで日本語を使うための設定として、サーバーのコード系をUTF-8に設定します。

この目的で以下のchaset.cnfを用意して、Dockerコンテナの／etc／mysql／conf.d／charset.cnfにCOPYします。

view plainprint?

1. [mysqld]  
2. character-set-server = utf8  

[mysqld]
character-set-server = utf8

entrypoint.sh

mysql-java-embulk-dockerのentrypoint.shは以下になります。

view plainprint?

1. #!/bin/bash  
2.   
3. # WAIT_DB_TIMER  
4. # WAIT_CONTAINER_KEY  
5.   
6. # set -x  
7.   
8. set -e  
9.   
10. echo Wait contaner key: ${WAIT_CONTAINER_KEY:=mysql-java-embulk-docker}  
11. echo Redis host: ${REDIS_SERVER_HOST:=$REDIS_PORT_6379_TCP_ADDR}  
12. echo Redis port: ${REDIS_SERVER_PORT:=$REDIS_PORT_6379_TCP_PORT}  
13.   
14. function check_db {  
15.     if [ "$MYSQL_ROOT_PASSWORD" ]; then  
16.  mysql -u root -p"$MYSQL_ROOT_PASSWORD" -e "status"  
17.     elif [ "$MYSQL_ALLOW_EMPTY_PASSWORD" ]; then  
18.  mysql -e "status"  
19.     else  
20.  exit 1  
21.     fi  
22. }  
23.   
24. function wait_db {  
25.     result=1  
26.     for i in $(seq 1 ${WAIT_DB_TIMER:-10})  
27.     do  
28.  sleep 1s  
29.  result=0  
30.  check_db && break  
31.  result=1  
32.     done  
33.     if [ $result = 1 ]; then  
34.  exit 1  
35.     fi  
36. }  
37.   
38. if [ "${1:0:1}" = '-' ]; then  
39.     set -- mysqld "$@"  
40. fi  
41.   
42. is_install=false  
43.   
44. if [ "$1" = 'mysqld' ]; then  
45.     # read DATADIR from the MySQL config  
46.     DATADIR="$("$@" --verbose --help 2>/dev/null | awk '$1 == "datadir" { print $2; exit }')"  
47.   
48.     if [ ! -d "$DATADIR/mysql" ]; then  
49.         if [ -z "$MYSQL_ROOT_PASSWORD" -a -z "$MYSQL_ALLOW_EMPTY_PASSWORD" ]; then  
50.             echo >&2 'error: database is uninitialized and MYSQL_ROOT_PASSWORD not set'  
51.             echo >&2 '  Did you forget to add -e MYSQL_ROOT_PASSWORD=... ?'  
52.             exit 1  
53.         fi  
54.   
55.  is_install=true  
56.   
57.         echo 'Running mysql_install_db ...'  
58.         mysql_install_db --datadir="$DATADIR"  
59.         echo 'Finished mysql_install_db'  
60.   
61.         # These statements _must_ be on individual lines, and _must_ end with  
62.         # semicolons (no line breaks or comments are permitted).  
63.         # TODO proper SQL escaping on ALL the things D:  
64.   
65.         tempSqlFile='/tmp/mysql-first-time.sql'  
66.         cat > "$tempSqlFile" <<-EOSQL  
67.             DELETE FROM mysql.user ;  
68.             CREATE USER 'root'@'%' IDENTIFIED BY '${MYSQL_ROOT_PASSWORD}' ;  
69.             GRANT ALL ON *.* TO 'root'@'%' WITH GRANT OPTION ;  
70.             DROP DATABASE IF EXISTS test ;  
71. EOSQL  
72.   
73.         if [ "$MYSQL_DATABASE" ]; then  
74.             echo "CREATE DATABASE IF NOT EXISTS \`$MYSQL_DATABASE\` ;" >> "$tempSqlFile"  
75.         fi  
76.   
77.         if [ "$MYSQL_USER" -a "$MYSQL_PASSWORD" ]; then  
78.             echo "CREATE USER '$MYSQL_USER'@'%' IDENTIFIED BY '$MYSQL_PASSWORD' ;" >> "$tempSqlFile"  
79.   
80.             if [ "$MYSQL_DATABASE" ]; then  
81.                 echo "GRANT ALL ON \`$MYSQL_DATABASE\`.* TO '$MYSQL_USER'@'%' ;" >> "$tempSqlFile"  
82.             fi  
83.         fi  
84.   
85.         echo 'FLUSH PRIVILEGES ;' >> "$tempSqlFile"  
86.   
87.  # http://qiita.com/toritori0318/items/242274d4f5794e2f68e5  
88.         # setup  
89.         echo "use $MYSQL_DATABASE;" >> "$tempSqlFile"  
90.  if [ -e "/opt/setup.d/setup.sql"]; then  
91.             cat /opt/setup.d/setup.sql >> "$tempSqlFile"  
92.  fi  
93.         # start mysql  
94.         set -- "$@" --init-file="$tempSqlFile"  
95.     fi  
96.   
97.     chown -R mysql:mysql "$DATADIR"  
98. fi  
99.   
100. exec "$@" &  
101.   
102. wait_db  
103.   
104. if [ -e "/opt/setup.d/setup.yml" ]; then  
105.     if [ $is_install=true ]; then  
106.  echo "embulk run setup.yml"  
107.  cd /opt/setup.d && /opt/embulk run setup.yml  
108.     fi  
109. fi  
110.   
111. if [ -n "$REDIS_SERVER_HOST" ]; then  
112.     redis-cli -h $REDIS_SERVER_HOST -p$REDIS_SERVER_PORT SET $WAIT_CONTAINER_KEY up  
113. fi  
114.   
115. sleep infinity  

#!/bin/bash

# WAIT_DB_TIMER
# WAIT_CONTAINER_KEY

# set -x

set -e

echo Wait contaner key: ${WAIT_CONTAINER_KEY:=mysql-java-embulk-docker}
echo Redis host: ${REDIS_SERVER_HOST:=$REDIS_PORT_6379_TCP_ADDR}
echo Redis port: ${REDIS_SERVER_PORT:=$REDIS_PORT_6379_TCP_PORT}

function check_db {
    if [ "$MYSQL_ROOT_PASSWORD" ]; then
 mysql -u root -p"$MYSQL_ROOT_PASSWORD" -e "status"
    elif [ "$MYSQL_ALLOW_EMPTY_PASSWORD" ]; then
 mysql -e "status"
    else
 exit 1
    fi
}

function wait_db {
    result=1
    for i in $(seq 1 ${WAIT_DB_TIMER:-10})
    do
 sleep 1s
 result=0
 check_db && break
 result=1
    done
    if [ $result = 1 ]; then
 exit 1
    fi
}

if [ "${1:0:1}" = '-' ]; then
    set -- mysqld "$@"
fi

is_install=false

if [ "$1" = 'mysqld' ]; then
    # read DATADIR from the MySQL config
    DATADIR="$("$@" --verbose --help 2>/dev/null | awk '$1 == "datadir" { print $2; exit }')"

    if [ ! -d "$DATADIR/mysql" ]; then
        if [ -z "$MYSQL_ROOT_PASSWORD" -a -z "$MYSQL_ALLOW_EMPTY_PASSWORD" ]; then
            echo >&2 'error: database is uninitialized and MYSQL_ROOT_PASSWORD not set'
            echo >&2 '  Did you forget to add -e MYSQL_ROOT_PASSWORD=... ?'
            exit 1
        fi

 is_install=true

        echo 'Running mysql_install_db ...'
        mysql_install_db --datadir="$DATADIR"
        echo 'Finished mysql_install_db'

        # These statements _must_ be on individual lines, and _must_ end with
        # semicolons (no line breaks or comments are permitted).
        # TODO proper SQL escaping on ALL the things D:

        tempSqlFile='/tmp/mysql-first-time.sql'
        cat > "$tempSqlFile" <<-EOSQL
            DELETE FROM mysql.user ;
            CREATE USER 'root'@'%' IDENTIFIED BY '${MYSQL_ROOT_PASSWORD}' ;
            GRANT ALL ON *.* TO 'root'@'%' WITH GRANT OPTION ;
            DROP DATABASE IF EXISTS test ;
EOSQL

        if [ "$MYSQL_DATABASE" ]; then
            echo "CREATE DATABASE IF NOT EXISTS \`$MYSQL_DATABASE\` ;" >> "$tempSqlFile"
        fi

        if [ "$MYSQL_USER" -a "$MYSQL_PASSWORD" ]; then
            echo "CREATE USER '$MYSQL_USER'@'%' IDENTIFIED BY '$MYSQL_PASSWORD' ;" >> "$tempSqlFile"

            if [ "$MYSQL_DATABASE" ]; then
                echo "GRANT ALL ON \`$MYSQL_DATABASE\`.* TO '$MYSQL_USER'@'%' ;" >> "$tempSqlFile"
            fi
        fi

        echo 'FLUSH PRIVILEGES ;' >> "$tempSqlFile"

 # http://qiita.com/toritori0318/items/242274d4f5794e2f68e5
        # setup
        echo "use $MYSQL_DATABASE;" >> "$tempSqlFile"
 if [ -e "/opt/setup.d/setup.sql"]; then
            cat /opt/setup.d/setup.sql >> "$tempSqlFile"
 fi
        # start mysql
        set -- "$@" --init-file="$tempSqlFile"
    fi

    chown -R mysql:mysql "$DATADIR"
fi

exec "$@" &

wait_db

if [ -e "/opt/setup.d/setup.yml" ]; then
    if [ $is_install=true ]; then
 echo "embulk run setup.yml"
 cd /opt/setup.d && /opt/embulk run setup.yml
    fi
fi

if [ -n "$REDIS_SERVER_HOST" ]; then
    redis-cli -h $REDIS_SERVER_HOST -p $REDIS_SERVER_PORT SET $WAIT_CONTAINER_KEY up
fi

sleep infinity

MySQL公式をベースに、データ投入用SQLおよびEmbulkでデータ投入するように拡張したものです。

setup.sql

／opt／setup.d／setup.sqlとしてデータ投入用SQLが存在する場合は、MySQLの初期起動スクリプトにこの内容を追加することで、起動時にデータ投入されるようになっています。

／opt／setup.dはDockerfileでVolumeなっていて、外部からディレクトリをマウントして使用することを想定しています。

データ投入用SQLは「Docker公式のmysqlイメージを使いつつ初期データも投入する」の記事を参考にしました。

setup.yml

／opt／setup.d／setup.ymlとしてデータ投入用Embulk記述ファイルが存在する場合は、Embulkを使ってデータ投入するようになっています。

ただし、MySQLの起動が完了した後でないとデータ投入ができないのでmysqlコマンドを使って待ち合わせ処理を行っています。

Redisによる同期

mysql-java-embulk-dockerコンテナの起動終了の待ち合わせのためRedisを使用します。

具体的には以下のように、外部コンテナでRedisが起動されている場合に、redis-cliコマンドを使って環境変数WAIT＿CONTAINER＿KEYで指定されたスロットに「up」という文字列を設定しています。

view plainprint?

1. if [ -n "$REDIS_SERVER_HOST" ]; then  
2.     redis-cli -h $REDIS_SERVER_HOST -p$REDIS_SERVER_PORT SET $WAIT_CONTAINER_KEY up  
3. fi  

if [ -n "$REDIS_SERVER_HOST" ]; then
    redis-cli -h $REDIS_SERVER_HOST -p $REDIS_SERVER_PORT SET $WAIT_CONTAINER_KEY up
fi

アプリケーション側は、Redisのこのスロットがupになるまでポーリングで待ち合わせることで同期を取ることになります。

終了抑止

最後にDockerでサービスを記述する時のお約束としてsleepコマンドで終了抑止を行っています。

Docker Hub

Docker HubはGitHubやBitBucketと連動した自動ビルド機能を提供しています。

mysql-java-embulk-dockerもこの設定を行っているので、以下の場所にDockerイメージが自動ビルドされます。

• https://registry.hub.docker.com/u/asami/mysql-java-embulk-docker/

このイメージは「asami/mysql-java-embulk-docker」という名前で利用することができます。

サンプル

Dockerイメージ「asami/mysql-java-embulk-docker」を利用してテストデータの投入を行うサンプルを作ってみます。

サンプルのコードはGitHubのmysql-java-embulk-dockerのsampleディレクトリにあるので、詳細はこちらを参照して下さい。

• https://github.com/asami/mysql-java-embulk-docker/tree/master/sample

docker-compose.yml

サンプルプログラムのdocker-compose.ymlは以下になります。

view plainprint?

1. app:  
2.   build: .  
3.   links:  
4.     - mysql  
5.     - redis  
6.   environment:  
7.     WAIT_CONTAINER_KEY: mysql-java-embulk-docker  
8.     MYSQL_SERVER_USER: baseball  
9.     MYSQL_SERVER_PASSWORD: baseball  
10. mysql:  
11.   image: asami/mysql-java-embulk-docker  
12.   links:  
13.     - redis  
14.   ports:  
15.     - ":3306"  
16.   volumes:  
17.     - setup.d:/opt/setup.d  
18.   environment:  
19.     MYSQL_USER: baseball  
20.     MYSQL_PASSWORD: baseball  
21.     MYSQL_ROOT_PASSWORD: baseball  
22.     MYSQL_DATABASE: baseball  
23. redis:  
24.   image: redis  
25.   ports:  
26.     - ":6379"  

app:
  build: .
  links:
    - mysql
    - redis
  environment:
    WAIT_CONTAINER_KEY: mysql-java-embulk-docker
    MYSQL_SERVER_USER: baseball
    MYSQL_SERVER_PASSWORD: baseball
mysql:
  image: asami/mysql-java-embulk-docker
  links:
    - redis
  ports:
    - ":3306"
  volumes:
    - setup.d:/opt/setup.d
  environment:
    MYSQL_USER: baseball
    MYSQL_PASSWORD: baseball
    MYSQL_ROOT_PASSWORD: baseball
    MYSQL_DATABASE: baseball
redis:
  image: redis
  ports:
    - ":6379"

setup.dをコンテナの／opt／setup.dにマウントしています。

setup.dには後述のsetup.ymlとデータファイルBatting.csvが格納されています。

Batting.csvは以下のサイトからデータを取得しました。

• http://seanlahman.com/baseball-archive/statistics/

setup.yml

setup.ymlはembulkで移入するデータの情報を記述したものです。

view plainprint?

1. in:  
2.   type: file  
3.   path_prefix: Batting.csv  
4.   parser:  
5.     charset: UTF-8  
6.     newline: CRLF  
7.     type: csv  
8.     delimiter: ','  
9.     quote: '"'  
10.     escape: ''  
11.     skip_header_lines: 1  
12.     columns:  
13.     - {name: playerID, type: string}  
14.     - {name: yearID, type: long}  
15.     - {name: stint, type: long}  
16.     - {name: teamID, type: string}  
17.     - {name: lgID, type: string}  
18.     - {name: G, type: long}  
19.     - {name: AB, type: long}  
20.     - {name: R, type: long}  
21.     - {name: H, type: long}  
22.     - {name: 2B, type: long}  
23.     - {name: 3B, type: long}  
24.     - {name: HR, type: long}  
25.     - {name: RBI, type: long}  
26.     - {name: SB, type: long}  
27.     - {name: CS, type: long}  
28.     - {name: BB, type: long}  
29.     - {name: SO, type: long}  
30.     - {name: IBB, type: long}  
31.     - {name: HBP, type: long}  
32.     - {name: SH, type: long}  
33.     - {name: SF, type: long}  
34.     - {name: GIDP, type: long}  
35. exec: {}  
36. out:  
37.   type: mysql  
38.   host: localhost  
39.   database: baseball  
40.   user: baseball  
41.   password: baseball  
42.   table: batting  
43.   mode: insert  

in:
  type: file
  path_prefix: Batting.csv
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: ','
    quote: '"'
    escape: ''
    skip_header_lines: 1
    columns:
    - {name: playerID, type: string}
    - {name: yearID, type: long}
    - {name: stint, type: long}
    - {name: teamID, type: string}
    - {name: lgID, type: string}
    - {name: G, type: long}
    - {name: AB, type: long}
    - {name: R, type: long}
    - {name: H, type: long}
    - {name: 2B, type: long}
    - {name: 3B, type: long}
    - {name: HR, type: long}
    - {name: RBI, type: long}
    - {name: SB, type: long}
    - {name: CS, type: long}
    - {name: BB, type: long}
    - {name: SO, type: long}
    - {name: IBB, type: long}
    - {name: HBP, type: long}
    - {name: SH, type: long}
    - {name: SF, type: long}
    - {name: GIDP, type: long}
exec: {}
out:
  type: mysql
  host: localhost
  database: baseball
  user: baseball
  password: baseball
  table: batting
  mode: insert

CSVファイルから入力したデータをMySQLに投入する際の標準的な指定と思います。

Dockerfile

サンプルアプリケーションのDockerfileは以下になります。

view plainprint?

1. FROM mysql  
2. MAINTAINER asami  
3.   
4. ENV MYSQL_ALLOW_EMPTY_PASSWORD true  
5.   
6. RUN apt-get update && apt-get -y install redis-server  
7.   
8. COPY app.sh /opt/app.sh  
9. RUN chmod +x /opt/app.sh  
10. ADD https://raw.githubusercontent.com/asami/mysql-java-embulk-docker/master/lib/mysql-java-embulk-docker-lib.sh /opt/mysql-java-embulk-docker-lib.sh  
11.   
12. ENTRYPOINT /opt/app.sh  

FROM mysql
MAINTAINER asami

ENV MYSQL_ALLOW_EMPTY_PASSWORD true

RUN apt-get update && apt-get -y install redis-server

COPY app.sh /opt/app.sh
RUN chmod +x /opt/app.sh
ADD https://raw.githubusercontent.com/asami/mysql-java-embulk-docker/master/lib/mysql-java-embulk-docker-lib.sh /opt/mysql-java-embulk-docker-lib.sh

ENTRYPOINT /opt/app.sh

アプリケーションでmysqlコマンドを使うので、MySQL公式Dockerイメージをベースにしました。

mysql-java-embulk-dockerコンテナとの同期にRedisを使うのでRedisをインストールしています。

また、アプリケーション起動シェルの共通ライブラリmysql-java-embulk-docker-lib.shをGitHubからコンテナ内にコピーしています。

アプリケーション起動の動きは大きく以下の3つの部分に分かれます。

• パラメタの取り込み
• データ投入の待ち合わせ
• アプリケーションロジック

この中の「パラメタの取り込み」と「データ投入の待ち合わせ」をmysql-java-embulk-docker-lib.shが行います。

app.sh

サンプルアプリケーションの実行スクリプトapp.shは以下になります。

view plainprint?

1. #! /bin/bash  
2.   
3. # set -x  
4.   
5. set -e  
6.   
7. DIR="${BASH_SOURCE%/*}"  
8. if [[ ! -d "$DIR" ]]; then DIR="$PWD"; fi  
9. source $DIR/mysql-java-embulk-docker-lib.sh  
10.   
11. mysql -u $MYSQL_SERVER_USER -p$MYSQL_SERVER_PASSWORD --host=$MYSQL_SERVER_HOST --port=$MYSQL_SERVER_PORT -e "select count(*) from baseball.batting"  

#! /bin/bash

# set -x

set -e

DIR="${BASH_SOURCE%/*}"
if [[ ! -d "$DIR" ]]; then DIR="$PWD"; fi
source $DIR/mysql-java-embulk-docker-lib.sh

mysql -u $MYSQL_SERVER_USER -p$MYSQL_SERVER_PASSWORD --host=$MYSQL_SERVER_HOST --port=$MYSQL_SERVER_PORT -e "select count(*) from baseball.batting"

まず、共通ライブラリmysql-java-embulk-docker-lib.shをsourceで取り込んでいます。

この中でRedisを使った同期が行われ、Embulkによるデータ投入が完了した状態でアプリケーションロジックに入ってきます。

今回のアプリケーションロジックは非常に単純で以下の処理を行います。

• baseball.battingテーブルの総レコード数を取得する

この問合せ処理をmysqlコマンドを使って行っています。

実行

docker-composeのbuildコマンドでビルドします。

$ docker-compose build

docker-composeのupコマンドでビルドします。

$ docker-compose up

動作過程がコンソールに出力されますが、最後の方で以下のような出力があります。

app_1   | count(*)
app_1   | 99846

無事、Batting.csvをMySQLのbaseball.battingテーブルに投入した後、baseball.battingテーブルの総レコード数を取得することができました。　

まとめ

SQLとEmbulkを使ってデータ投入できるMySQL用のDockerコンテナを作ってみました。

アプリケーション開発では、テスト用データベースの準備とデータ投入が大きな手間であり、テスト自動化の障壁にもなっていたので、今回開発したDockerイメージをアプリケーション開発に適用していきたいと思います。

それにしても、Dockerのイメージ開発はシェルスクリプトプログラミングということを実感しました。相当錆び付いていましたが、なんとか動くものができました。

諸元

• Mac OS 10.7.5
• docker 1.6
• docker-compose 1.2.0

[docker] Docker ComposeでMySQLを使う

製品の評価や、開発時のテストなどDBをカジュアルに作成して消したいシチュエーションは多々あります。

開発マシンにDBを入れてデータ投入する運用だと、開発マシン環境が汚れてきたり、復数の設定が共存して収集がつかなくなったり、DBやライブラリのバージョンに依存する場合の切り替えに困ったりということになりがちです。

そこで以前だとVMを使ったり、AWSのようなクラウド上にDBを立てたりしていたわけですが、設定はそのものが難しくはないものの、毎回環境を構築する作業が必要になったりと、まだまだ手間のかかる作業でした。

言うまでもなく、このような問題がDockerで一気に解決したのは大変画期的なことです。

今回はDockerを使ってDBをカジュアルに構築する方法について考えてみます。

SoS JobScheduler

今回はジョブ管理製品のSoS JobSchedulerを使ってみました。

• http://www.sos-berlin.com/jobscheduler

JobSchedulerはバックエンドにDBを使っているので、このDBをDocker上でどのようにして構築して接続するのかという点がポイントです。

またJobSchedulerはapt-getやrpmといったインストーラに対応しておらずインストールが手作業になります。このような製品の配布方式はけっこうありますが、この部分をDockerfile側、起動スクリプト側双方のシェルスクリプトで対応することになります。

MySQL公式Dockerイメージ

MySQLをインストールしたマイDockerイメージを作る方法も有力ですが、MySQLの公式Dockerイメージが機能豊富で結構便利だったので使ってみました。

また、JobSchedulerとMySQLを同じDockerイメージにインストールする方法もありますが、JobScheduler用Dockerイメージとしての汎用性が損なわれるので、避けたほうがよいでしょう。

このため、今回使用するDockerイメージはMySQL公式イメージとJobSchedulerをインストールしたマイイメージの2つになります。このように復数のイメージを接続する場合、dockerコマンドののパラメタを設定する方法もありますが、かなり煩雑です。

この問題に対応するため今回はDocker Composeを使ってみました。

Docker Compose

Docker Compose(旧fig)は復数のDockerイメージを連携動作させるための機能です。

• https://docs.docker.com/compose/

定義ファイルdocker-compose.ymlの設定に従って、復数のDockerコンテナを同時に立ち上げ、Dockerのlinking systemを用いて各Dockerコンテナをリンクで接続する処理を行います。

設定

それでは、Docker Composeを用いた設定を行います。設定結果の全体はGitHubにありますので必要に応じて参照して下さい。

• https://github.com/asami/SoS-JobScheduler-docker

docker-compose.yml

docker-compose.ymlの設定は以下になります。

view plainprint?

1. jobscheduler:  
2.   build: .  
3.   links:  
4.     - db  
5.   ports:  
6.     - "4444:4444"  
7. db:  
8.   image: mysql  
9.   ports:  
10.     - ":3306"  
11.   volumes:  
12.     - conf.d/etc.mysql.conf.d:/etc/mysql/conf.d  
13.   environment:  
14.     MYSQL_USER: jobscheduler  
15.     MYSQL_PASSWORD: jobscheduler  
16.     MYSQL_ROOT_PASSWORD: jobscheduler  
17.     MYSQL_DATABASE: jobscheduler  

jobscheduler:
  build: .
  links:
    - db
  ports:
    - "4444:4444"
db:
  image: mysql
  ports:
    - ":3306"
  volumes:
    - conf.d/etc.mysql.conf.d:/etc/mysql/conf.d
  environment:
    MYSQL_USER: jobscheduler
    MYSQL_PASSWORD: jobscheduler
    MYSQL_ROOT_PASSWORD: jobscheduler
    MYSQL_DATABASE: jobscheduler

jobschedulerとdbの2つのDockerコンテナの定義をしています。

jobschedulerコンテナは自前のDockerfileを使ったコンテナです。linksでdbコンテナをリンクする設定を行っています。

dbコンテナはMySQL公式イメージをそのまま使っています。ポイントとなるのはenvironmentで指定している4つの環境変数です。これらの環境変数を設定することで、自動的に必要な初期設定をしてくれるようになっています。

多くのケースで、MySQL公式イメージが提供している環境変数で目的が足りると思われるので、MySQL公式イメージはかなり使い出がありそうです。

Dockerfile

JobSchedulerを実行するDockerfileの設定は以下になります。

view plainprint?

1. FROM dockerfile/java:oracle-java8  
2. MAINTAINER asami  
3.   
4. ENV JOBSCHEDULER_VERSION 1.9.0  
5.   
6. RUN mkdir -p /opt/jobscheduler && cd /opt/jobscheduler; curl -L http://freefr.dl.sourceforge.net/project/jobscheduler/jobscheduler_linux-x64.$JOBSCHEDULER_VERSION.tar.gz -o - | tar -xz --strip-components=1  
7.   
8. # SSH, API/HTTP, API/HTTPS, JOC  
9. EXPOSE 22 44440 8443 4444  
10.   
11. USER root  
12.   
13. COPY scheduler_install.xml /opt/jobscheduler/scheduler_install.xml  
14.   
15. # Set the default command to run when starting the container  
16. COPY startup-jobscheduler.sh /opt/startup-jobscheduler.sh  
17. CMD ["/opt/startup-jobscheduler.sh"]  

FROM dockerfile/java:oracle-java8
MAINTAINER asami

ENV JOBSCHEDULER_VERSION 1.9.0

RUN mkdir -p /opt/jobscheduler && cd /opt/jobscheduler; curl -L http://freefr.dl.sourceforge.net/project/jobscheduler/jobscheduler_linux-x64.$JOBSCHEDULER_VERSION.tar.gz -o - | tar -xz --strip-components=1

# SSH, API/HTTP, API/HTTPS, JOC
EXPOSE 22 44440 8443 4444

USER root

COPY scheduler_install.xml /opt/jobscheduler/scheduler_install.xml

# Set the default command to run when starting the container
COPY startup-jobscheduler.sh /opt/startup-jobscheduler.sh
CMD ["/opt/startup-jobscheduler.sh"]

JobSchedulerがJava 8依存なので、基盤イメージとしてjava8版のJava公式イメージを指定しています。

wgetはインストールが必要なので用いずcurlを使っています。curlとtarを連動させている以下の行は:

view plainprint?

1. RUN mkdir -p /opt/jobscheduler && cd /opt/jobscheduler; curl -L http://freefr.dl.sourceforge.net/project/jobscheduler/jobscheduler_linux-x64.$JOBSCHEDULER_VERSION.tar.gz -o - | tar -xz --strip-components=1  

RUN mkdir -p /opt/jobscheduler && cd /opt/jobscheduler; curl -L http://freefr.dl.sourceforge.net/project/jobscheduler/jobscheduler_linux-x64.$JOBSCHEDULER_VERSION.tar.gz -o - | tar -xz --strip-components=1

• 中間ファイルを残さない。
• 配布アーカイブにあるディレクトリ"jobscheduler.1.9.0"を"jobscheduler"に付け替え。こうすることで、後続の処理でバージョン番号を意識する処理を減らすことができます。

という意図です。このようなケースのイディオム的なスクリプトです。

startup-jobscheduler.sh

dockerの定義で比較的難しいのは起動スクリプトのところです。

基本的には、ターゲットのプログラムを起動するだけなのですが、環境との整合性を取るための処理を色々と書く必要があります。

今回の起動スクリプトはstartup-jobscheduler.shで、DockerfileのCMDで指定しています。

view plainprint?

1. #! /bin/bash  
2.   
3. sleep 10s  
4.   
5. sed -i -e "s/{{DB_PORT_3306_TCP_ADDR}}/$DB_PORT_3306_TCP_ADDR/g" /opt/jobscheduler/scheduler_install.xml  
6. sed -i -e "s/{{DB_PORT_3306_TCP_PORT}}/$DB_PORT_3306_TCP_PORT/g" /opt/jobscheduler/scheduler_install.xml  
7.   
8. (cd /opt/jobscheduler;/usr/bin/java -jar jobscheduler_linux-x64.$JOBSCHEDULER_VERSION.jar scheduler_install.xml)  
9.   
10. sleep infinity  

#! /bin/bash

sleep 10s

sed -i -e "s/{{DB_PORT_3306_TCP_ADDR}}/$DB_PORT_3306_TCP_ADDR/g" /opt/jobscheduler/scheduler_install.xml
sed -i -e "s/{{DB_PORT_3306_TCP_PORT}}/$DB_PORT_3306_TCP_PORT/g" /opt/jobscheduler/scheduler_install.xml

(cd /opt/jobscheduler;/usr/bin/java -jar jobscheduler_linux-x64.$JOBSCHEDULER_VERSION.jar scheduler_install.xml)

sleep infinity

DBの起動を待ち合わせるためsleepコマンドで10秒ウエイト入れています。

scheduler_install.xmlの設定を、Docker実行時の環境に適合するようにsedコマンドで書き換えています。環境変数DB＿PORT＿3306＿TCP＿ADDRとDB＿PORT＿3306＿TCP＿PORTはDockerのlinking systemが設定してくる環境情報です。これらの情報の取り込みを起動スクリプトで対応する必要があります。

JobSchedulerは起動後、自動的にバックグラウンドになってしまうため、そのままstartup-jobscheduler.shが終わるとそのままDockerも終わってしまいます。そこでsleepコマンドで永久にウエイトするようにしています。

実行

Docker Composeを使ってJobSchedulerを立ち上げてみましょう。

設定の取得

GitHubから設定一式を取得します。

$ git clone https://github.com/asami/SoS-JobScheduler-docker.git

ビルド

docker-composeコマンドのbuildを実行します。

$ docker-compose build

実行

docker-composeコマンドのupを実行します。

$ docker-compose up

起動は以上で終了です。

以下のアドレスにアクセスするとJobSchedulerの管理画面が表示されます。(Macの場合)

• http://192.168.59.103:4444/

まとめ

Docker Composeを使って、自前のJobSchedulerコンテナとMySQL公式イメージを連動させ、JobSchedulerの実行環境を作ってみました。

MySQL公式イメージの機能が結構豊富なので、DB側は設定だけで使用することができました。

Dockerも便利ですが、Docker Composeもかなり便利で、アイデア次第で色々と応用がありそうです。

諸元

Mac OS 10.7.5 docker 1.6 docker-compose 1.2.0