sparklyrでExploratoryからSparkに接続してみる(dplyr編)

ビッグデータ分析基盤として、もはや定番となっているApache Spark。

前回はSQLでSparkにSQLクエリを投げてデータ分析をする例をご説明しましたが、今回はdplyrのクエリを使った例をご紹介します。

Sparkにアメリカの航空会社の遅延データがあったとすると、実は以下のようにdplyrのクエリを投げてSparkからデータを取得できます。

flights <- tbl(sc, "flights") %>%
  group_by(CARRIER, ORIGIN_STATE_ABR) %>% 
  summarize(DEP_DELAY_AVG = mean(DEP_DELAY)) %>%
  top_n(10, DEP_DELAY_AVG) %>%
  collect()

Exploratoryのユーザー、もしくは今時のRのユーザーならdplyrの文法(データ・ラングリングのグラマーと言われているもの)に慣れていると思いますが、煩雑なSQLを手で書くのと比べて、格段に直感的にSparkのデータ分析ができます。

ということで、今日はExploratoryの中から、sparklyrを使ってSparkにdplyrクエリーを投げて、その結果だけをスマートに取ってくるにはどうするかということを紹介をしたいと思います。

準備

まずは最初に、sparklyrというRパッケージと、Javaをインストールします。

sparklyrパッケージのインストール

ExploratoryからSparkに接続するには、sparklyrというRから直接SparkにアクセスするためのRのパッケージをインストールする必要があります。 Exploratoryの中で、CRANからsparklyrパッケージを以下のように直接インストールします。

Javaのインストール

お使いのマシンにJavaがインストールされてない場合は、こちらからダウンロードしてインストールしてください。

Sparkからdplyrでデータを読み込む

次に、Rスクリプトを使ったデータソースでSparkにsparklyrを使って接続します。

Rスクリプトの中で、次のようにSparkに接続します。(ここではlocalにデプロイしたSparkを例として取り上げています。)

# Sparkに接続 (ローカルのマシンの例)
# 環境変数の設定
Sys.setenv(SPARK_HOME = "/Users/hidekoji/spark/spark-2.2.0-bin-hadoop2.7")
# sparklyrを読み込む
library(sparklyr)
# Sparkに接続する
sc <- spark_connect(master = "spark://Hidetakas-MacBook-Pro.local:7077")

さて、各航空会社毎に平均出発遅延のトップ10の州を計算してみましょう。dplyrのクエリは以下の通りです。。

flights <- tbl(sc, "flights") %>%
  group_by(CARRIER, ORIGIN_STATE_ABR) %>% 
  summarize(DEP_DELAY_AVG = mean(DEP_DELAY)) %>%
  top_n(10, DEP_DELAY_AVG) %>%
  collect()

最後のcollect()はクエリの実行結果をデータフレームにするための関数です。この記事の一番最後にこの2つのセクションをまとめたスクリプトがありますので、コピー & ペーストとして、必要な部分を変更して試せます。

データの取得ボタンをクリックして結果を確認し、保存ボタンをクリックし、データフレームとして保存します。

データを保存して、テーブルビューで結果を確認します。

このdplyrのクエリの結果をバー・チャートにして、「繰り返し」に航空会社を割り当てると、航空会社毎の出発遅延時間Top10の州を可視化できます。

まとめ

sparklyrを使うと、このようにSparkに接続して、ScalaやPythonでプログラミングしなくても、そしてSQLのクエリさえも書かなくても、dplyrクエリでデータを取得し可視化やさらなる分析が可能になります。

データサイエンスを本格的に学んでみたいという方へ

今年10月に、Exploratory社がシリコンバレーで行っている研修プログラムを日本向けにした、データサイエンス・ブートキャンプの第3回目が東京で行われます。本格的に上記のようなデータサイエンスの手法を、プログラミングなしで学んでみたい方、そういった手法を日々のビジネスに活かしてみたい方はぜひこの機会に、参加を検討してみてはいかがでしょうか。こちらに詳しい情報がありますのでぜひご覧ください。

補足

この例を試すには

以下のスクリプトをRスクリプトデータソースとしてコピー & ペーストして、必要に応じて変更後実行してください。

# Sparkに接続 (ローカルのマシンの例)
# 環境変数の設定
Sys.setenv(SPARK_HOME = "/Users/hidekoji/spark/spark-2.2.0-bin-hadoop2.7")
# sparklyrを読み込む
library(sparklyr)
# Sparkに接続する
sc <- spark_connect(master = "spark://Hidetakas-MacBook-Pro.local:7077")
flights <- tbl(sc, "flights") %>%
  group_by(CARRIER, ORIGIN_STATE_ABR) %>% 
  summarize(DEP_DELAY_AVG = mean(DEP_DELAY)) %>%
  top_n(10, DEP_DELAY_AVG) %>%
  collect()