Loading [MathJax]/jax/output/CommonHTML/jax.js

2015年5月11日月曜日

[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にソースコードがありますので、詳細はこちらを参照して下さい。

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

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"]  

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−hREDIS_SERVER_HOST -p REDISSERVERPORTGETWAIT_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=COMMANDJARDIR/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  

基本的には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  

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

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

Docker Hub

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

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

サンプル

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

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

サンプルのコードはGitHubのspark-sql-scala-dockerの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. }  

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

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

view plainprint?
  1. val batting = SparkSqlUtils.createMysqlDataFrame("Simple Application", "batting")  
  2.   val count = batting.count()  
  3.   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)  

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

view plainprint?
  1. 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"  

自前では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は以下のサイトからデータを取得しました。

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
投稿者 時刻:
ラベル: , ,

0 件のコメント:

コメントを投稿

登録: コメントの投稿 (Atom)