astamuse Lab

astamuse Labとは、アスタミューゼのエンジニアとデザイナーのブログです。アスタミューゼの事業・サービスを支えている知識と舞台裏の今を発信しています。

Spark 2.0 で Spark SQLを試す

Spark Big Data データ開発エンジニア

こんにちは！Spark大好きな朴です。

本日はSpark 2.0で大幅の改善が行われてたSpark SQLについて書こうと思います。
弊社ではCDHのバージョンアップをこまめに行うことでSpark,HBaseなどのプロダクトの最新バージョンを常に試せる環境を作っております。

Spark 2.0についても先日弊社福田のもう待てない、Spark2.0の導入と実践にも書いてたとおり もう使えるようになりました。

ということで少し乗り遅れた感もありますが、本日はSpark 2.0でSpark SQLの実力を試したいと思います。

Spark 2.0でSpark SQLの主な変更点は以下の３つ

  • SparkSession
  • 性能改善
  • サポートするSQLが増えた

本日は上記３つの改善について触れてみたいと思います。

【変更その１】 SparkSQLのニューエントリポイントSparkSession

Spark 1.6まではユーザは以下のようにSqlContext(SparkSQL)とHiveContext(HiveQL)を使い分ける必要がありました。

// Spark SQL
val sc = new SparkContext(spConf)
val sqlContext = new SQLContext(sc)
sqlContext.sql("select * from hoge").show()

// HiveQL
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)
hiveContext.sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
hiveContext.sql("LOAD DATA LOCAL INPATH 'temp/kv1.txt' INTO TABLE src")
hiveContext.sql("FROM src SELECT key, value").collect().foreach(println)


Spark 2.0では以下のようSparkSessionでSparkSQL,HiveQL両方書けるようになりました。

val spark = SparkSession.builder()
      .master("yarn")
      .appName("Test")
      .getOrCreate()

// Spark SQL
spark.sql("select * from hoge").show()

// HiveQL
spark.sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
spark.sql("LOAD DATA LOCAL INPATH 'temp/kv1.txt' INTO TABLE src")
spark.sql("FROM src SELECT key, value").collect().foreach(println)

【変更その２】 性能面での改善

SQLおよびDataFrame関連の処理にてwhole stage code generationの新技術を使うことでおよそ2-10倍の性能改善を実現したそうです。
リリースノートの原文はこちら

それでは実際どうなのかちょっと試してみたいと思います。

以下同じ条件でSpark 1.6 と Spark 2.0でいくつか簡単なSQLを試してどれぐらい実行スピードが違うかを実験してみます。

データソースはwikipediaのデータを使います。 こちらのリンクよりenwiki-20170201-abstract.xmlをダウンロードします。 https://dumps.wikimedia.org/enwiki/20170201/

上記データは英語wikiの概要とタイトル、urlを含めたxmlになります。

<doc>
<title>Wikipedia: Anarchism</title>
<url>https://en.wikipedia.org/wiki/Anarchism</url>
<abstract>Anarchism is a political philosophy that advocates self-governed societies based on voluntary institutions.</abstract>
<links>
 <sublink linktype="nav"><anchor>External links</anchor><link>https://en.wikipedia.org/wiki/Anarchism#External_links</link></sublink>
</links>
</doc>
<doc>
 ..
 ..
</doc>


  • 前準備として上記のデータをhdfsにアップします。
hadoop fs -put enwiki-20170201-abstract.xml temp/


  • 以下のコマンドでspark2-shellを起動します。
spark2-shell --master yarn \
 --driver-memory 2G --executor-memory 4G  \
 --packages com.databricks:spark-xml_2.10:0.4.1 \
 --conf spark.serializer="org.apache.spark.serializer.KryoSerializer"

※注 spark2.0ではspark-xmlのバージョンを0.4.1以上に指定する必要があります。

  • XMLをDataFrameに変換します。
scala> val enwiki=spark.read.format("com.databricks.spark.xml").option("rowTag", "doc").option("inferSchema", "true").load("temp/enwiki-20170201-abstract.xml")
scala> enwiki.printSchema
root
 |-- abstract: string (nullable = true)
 |-- links: string (nullable = true)
 |-- title: string (nullable = true)
 |-- url: string (nullable = true)

scala> enwiki.createOrReplaceTempView("enwiki")


  • 実行時間計測用の関数を定義します。
scala > def showTiming[T](proc: => T): T = {
    val start=System.nanoTime()
    val res = proc
    val end = System.nanoTime()
    val esplatedSec = java.util.concurrent.TimeUnit.SECONDS.convert((end-start), java.util.concurrent.TimeUnit.NANOSECONDS)
    println("Time elapsed: " + esplatedSec + " secs")
    res
}


性能検証１ 件数count

Spark 2.0

scala> showTiming{ enwiki.count() }
Time elapsed: 116 secs
res4: Long = 5328417

結果でました。実行時間は116秒、件数は532万件ちょっとありました。

同じ処理をSpark 1.6で実行してみた結果


scala> showTiming{enwiki.count()}
Time elapsed: 146 secs
res4: Long = 5328417

こちらは実行時間が30秒増えてますね。 劇的ではありませんが、DataFrameの件数カウントではSpark2.0のほうが早いことが分かります。

性能検証２ like検索

Spark 2.0

scala> showTiming{ spark.sql("select count(*) from enwiki where abstract like '%Robot%'").show() }
+--------+
|count(1)|
+--------+
|     769|
+--------+

Time elapsed: 100 secs

Spark 2.0でlike検索の実行時間は116秒、件数は769件ありました

同じ処理をSpark 1.6で実行してみた結果


scala> showTiming{ sqlContext.sql("select count(*) from enwiki where abstract like '%Robot%'").show() }
+---+
|_c0|
+---+
|769|
+---+

Time elapsed: 143 secs

こちらは実行時間が43秒増えてますね。countと同じくSpark2.0のほうが性能的には上ですね

性能検証３ join処理

join系の処理を試してみたいと思います。 enwikiとjoinするためのデータを準備するためにsimplewiki-20170201-abstract.xmlを 以下のリンクよりダウンロードします。 https://dumps.wikimedia.org/enwiki/20170201/

データの構造はenwikiと同じでurlが違うだけです。

<doc>
<title>Wikipedia: April</title>
<url>https://simple.wikipedia.org/wiki/April</url>
<abstract>April is the fourth month of the year, and comes between March and May. It is one of four months to have 30 days.</abstract>
<links>
<sublink linktype="nav"><anchor>The Month</anchor><link>https://simple.wikipedia.org/wiki/April#The_Month</link></sublink>
</links>
</doc>
<doc>
..
..
</doc>


  • データをhdfsに上げます
hadoop fs -put simplewiki-20170201-abstract.xml temp/


  • simplewikiのデータをロードしてDataFrameに変換します。

scala> val simplewiki=spark.read.format("com.databricks.spark.xml").option("rowTag", "doc").option("inferSchema", "true").load("temp/simplewiki-20170201-abstract.xml")

scala> simplewiki.printSchema
root
 |-- abstract: string (nullable = true)
 |-- links: string (nullable = true)
 |-- title: string (nullable = true)
 |-- url: string (nullable = true)

scala> simplewiki.createOrReplaceTempView("simplewiki")


  • 下記のSQLでenwikiとsimplewikiで同じタイトルのデータ同士でinner joinをします。

Spark 2.0
scala> showTiming{spark.sql("select count(*) from enwiki en inner join simplewiki simple on (en.title = simple.title)").show()}
+--------+
|count(1)|
+--------+
|  107550|
+--------+

Time elapsed: 128 secs

Spark 2.0では128秒かかりました。

Spark 1.6

scala> showTiming{sqlContext.sql("select count(*) from enwiki en inner join simplewiki simple on (en.title = simple.title)").show()}
+------+
|   _c0|
+------+
|107550|
+------+

Time elapsed: 152 secs

Spark 1.6では152秒となり、Spark2.0より遅いことが分かります。

とりあえず、以上で簡単にcount,like検索,joinでSpark 2.0とSpark 1.6でSQLの性能検証を実施してみました。 どの処理もSpark 2.0で性能が改善されてることが分かりました。

【変更その３】 SQL2003をサポート、99個のTPC-DS クエリサポート

主には以下可能になりました。

  • ANSI-SQLとHive QLの両方をサポートするネイティブSQLパーサー
  • DDLが書けるようになった
  • Subqueryが書けるようになった
    • in,exists,not in, not existsなどなど（これは嬉しいー）
  • View canonicalization support (viewが使えるようになった？)

それではサブクエリ関連ちょっと触ってみたいと思います。

  • 給料を管理するテーブル(DataFrame)を定義
scala> val df1 = spark.createDataFrame(Array(("1001",1000),("1002",2000),("1003",1300),("1004",1500),("1005",1600),("1006",1700),("1007",1600),("1008",1700),("1009",1000),("1010",1900)))
df1: org.apache.spark.sql.DataFrame = [_1: string, _2: int]

scala> val salary=df1.withColumnRenamed("_1", "staffId").withColumnRenamed("_2", "salary")
salary: org.apache.spark.sql.DataFrame = [staffId: string, salary: int]

scala> salary.show()
+-------+------+
|staffId|salary|
+-------+------+
|   1001|  1000|
|   1002|  2000|
|   1003|  1300|
|   1004|  1500|
|   1005|  1600|
|   1006|  1700|
|   1007|  1600|
|   1008|  1700|
|   1009|  1000|
|   1010|  1900|
+-------+------+


  • 社員情報を管理するテーブル(DataFrame)を定義

scala> val df2 = spark.createDataFrame(Array(("1001","Staff1","Marketing"),("1002","Staff2","Marketing"),("1003","Staff3","Marketing"),("1004","Staff4","Sales"),("1005","Staff5","Sales"),("1006","Staff6","Sales"),("1007","Staff7","Sales"),("1008","Staff8","Development"),("1009","Staff9","Development"),("1010","Staff10","Development")))
df2: org.apache.spark.sql.DataFrame = [_1: string, _2: string ... 1 more field]

scala> val depart=df2.withColumnRenamed("_1", "staffId").withColumnRenamed("_2", "staffName").withColumnRenamed("_3", "depart")
depart: org.apache.spark.sql.DataFrame = [staffId: string, staffName: string ... 1 more field]

scala> depart.show
+-------+---------+-----------+
|staffId|staffName|     depart|
+-------+---------+-----------+
|   1001|   Staff1|  Marketing|
|   1002|   Staff2|  Marketing|
|   1003|   Staff3|  Marketing|
|   1004|   Staff4|      Sales|
|   1005|   Staff5|      Sales|
|   1006|   Staff6|      Sales|
|   1007|   Staff7|      Sales|
|   1008|   Staff8|Development|
|   1009|   Staff9|Development|
|   1010|  Staff10|Development|
+-------+---------+-----------+


  • (not) existsを試してみる
scala> salary.createOrReplaceTempView("salary")

scala> depart.createOrReplaceTempView("depart")

scala> spark.sql("select * from salary s where exists (select 1 from depart d where s.staffId = d.staffId and d.depart = 'Marketing')").show()

+-------+------+
|staffId|salary|
+-------+------+
|   1001|  1000|
|   1002|  2000|
|   1003|  1300|
+-------+------+

scala> spark.sql("select * from salary s where not exists (select 1 from depart d where s.staffId = d.staffId and d.depart = 'Marketing')").show()
+-------+------+
|staffId|salary|
+-------+------+
|   1004|  1500|
|   1005|  1600|
|   1006|  1700|
|   1007|  1600|
|   1008|  1700|
|   1009|  1000|
|   1010|  1900|
+-------+------+


  • (not) inを試してみる

scala> spark.sql("select * from salary s where staffId in (select staffId from depart d where d.depart = 'Marketing')").show()
+-------+------+
|staffId|salary|
+-------+------+
|   1001|  1000|
|   1002|  2000|
|   1003|  1300|
+-------+------+

scala> spark.sql("select * from salary s where staffId not in (select staffId from depart d where d.depart = 'Marketing')").show()
+-------+------+
|staffId|salary|
+-------+------+
|   1004|  1500|
|   1005|  1600|
|   1006|  1700|
|   1007|  1600|
|   1008|  1700|
|   1009|  1000|
|   1010|  1900|
+-------+------+

エラーなく実行できました！

終わりに

去年(2016年)の12月にSpark2.1がリリースされて、Spark2系も開発が活発に進められてる印象があります。 本日はSpark2系のSQL面での主な改善点を書きましたが、上記以外にもたくさんの改善が行われてますので、 これから既存のSparkジョブをSpark2系向けに移行しながらその良さを実感したいと思います。

ではまた！

Copyright © astamuse company, ltd. all rights reserved.

20160610170904