Scalaでのデータ分析。 21世紀の相関を考慮する

データ分析に適切なツールを選択することは非常に重要です。 国際的なデータサイエンスコンペティションを開催するKaggle.comフォーラムでは、どのツールが最適かをよく尋ねられます。 人気の最初の行は、RとPythonが占めています。 この記事では、Scalaプログラミング言語とSpark分散コンピューティングプラットフォームに基づいたデータ分析技術の代替スタックについて説明します。



どうやってこれに来たの？ Retail Rocketでは、非常に大きなデータセットで多くの機械学習を行っています。 以前は、バンドルのIPython + Pyhs2（Pythonのハイブドライバ）+ Pandas + Sklearnを使用してプロトタイプを開発しました。 2014年夏の終わりに、同じサーバーパークで生産性が3〜4倍向上することが実験により示されたため、Sparkに切り替えるという基本的な決定を行いました。



別のプラス-1つのプログラミング言語を使用して、バトルサーバーで動作するモデリングとコードを作成できます。 これは私たちにとって大きな利点でした。以前は、Hive、Pig、Java、Pythonの4つの言語を同時に使用していました。小さなチームではこれは深刻な問題です。



Sparkは、APIを介したPython / Scala / Javaとの連携をサポートします。 Scalaに書かれているのはSparkであるため、Scalaを選択することにしました。つまり、ソースコードを分析し、必要に応じてエラーを修正でき、さらにHadoop全体が実行されるJVMです。 Sparkプログラミング言語フォーラムの分析により、次のことがわかりました。



Scala：

+機能;

+ Sparkネイティブ;

+ JVMで動作します。つまり、Hadoopネイティブです。

+厳密な静的型付け;

-かなり複雑なエントリですが、コードは読み取り可能です。



Python：

+人気;

+シンプル;

-動的型付け;

-パフォーマンスはScalaよりも悪いです。



Java：

+人気;

+ Hadoopにネイティブ。

-コードが多すぎる。



Sparkのプログラミング言語の選択について詳しくは、 こちらをご覧ください 。



その時点でチームの誰もScalaを知らなかったため、選択は容易ではなかったと言わざるを得ません。

既知の事実：言語でうまくコミュニケーションをとることを学ぶには、言語環境に没頭し、できるだけ頻繁に使用する必要があります。 したがって、モデリングと迅速なデータ分析のために、Scalaを優先してPythonスタックを放棄しました。



まず、IPythonの代替品を見つける必要がありました。オプションは次のとおりでした。

1） Zeppelin -Spark用のIPythonのようなノートブック。

2）ISpark;

3）Spark Notebook;

4） IBMのSpark IPython Notebook 。



ISparkの選択は単純ですが、Scala / SparkのIPythonであるため、HighChartsとRグラフを比較的簡単に固定でき、Yarnクラスターに問題なく接続できました。



Scalaデータマイニング環境に関するストーリーは、3つの部分で構成されています。

1）ISparkのScalaでの簡単なタスク。Sparkでローカルに実行されます。

2）ISparkで動作するようにコンポーネントを構成およびインストールします。

3）ライブラリRを使用して、Scalaで機械学習タスクを作成します。

そして、この記事が人気がある場合は、他に2つ書いておきます。 ;）

挑戦する

質問に答えてみましょう：オンラインストアでの平均購入領収書は、地域、ブラウザの種類（モバイル/デスクトップ）、オペレーティングシステム、ブラウザのバージョンなど、クライアントの静的パラメータに依存しますか？ これは、 「相互情報」 （相互情報）を使用して実行できます。

Retail Rocketでは 、古典的なシャノンの公式、Kullback-Leiblerの発散、相互情報など、多くの場所で推奨アルゴリズムと分析にエントロピーを使用しています。 このトピックに関するRecSysカンファレンスでレポートを申請しました。 これらの対策は、マーフィーの有名な機械学習の教科書の小さなセクションではありますが、個別のセクションに当てられています。

Retail Rocketの実際のデータを分析してみましょう。 以前は、サンプルをクラスターからコンピューターにcsvファイルとしてコピーしました。

データの読み込み

ここでは、ローカルモードで起動されたISparkとSparkを使用します。つまり、すべての計算はローカルで行われ、分布はコアを通過します。 実際、すべてはコメントに書かれています。 最も重要なことは、出力でRDD（Sparkデータ構造）を取得することです。RDDは、コードで定義されているRow型のケースクラスのコレクションです。 これにより、_。categoryIdなどの「。」を介してフィールドにアクセスできます。



入り口で：

import org.apache.spark.rdd.RDD import org.apache.spark.sql._ import org.tribbloid.ispark.display.dsl._ import scala.util.Try val sqlContext = new org.apache.spark.sql.SQLContext(sc) import sqlContext.implicits._ //  CASE class,     dataframe case class Row(categoryId: Long, orderId: String ,cityId: String, osName: String, osFamily: String, uaType: String, uaName: String,aov: Double) //     val   sc (Spark Context),   Ipython  val aov = sc.textFile("file:///Users/rzykov/Downloads/AOVC.csv") //   val dataAov = aov.flatMap { line => Try { line.split(",") match { case Array(categoryId, orderId, cityId, osName, osFamily, uaType, uaName, aov) => Row(categoryId.toLong + 100, orderId, cityId, osName, osFamily, osFamily, uaType, aov.toDouble) } }.toOption }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

出力：

 MapPartitionsRDD[4] at map at <console>:28
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

それでは、データ自体を見てみましょう。



この行は、バージョン1.3.0でSparkに追加された新しいDataFrameデータ型を使用しています;これは、Pythonのpandasライブラリの類似した構造に非常に似ています。 toDfはRowケースクラスを取得します。これにより、フィールドの名前とタイプが取得されます。



さらに分析するには、任意の1つのカテゴリを選択する必要があります。できれば多くのデータを使用してください。 これを行うには、最も人気のあるカテゴリのリストを取得する必要があります。



入り口で：

 //   dataAov.map { x => x.categoryId } //   categoryId .countByValue() //     categoryId .toSeq .sortBy( - _._2) //       .take(10) //   10 
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

出力では、形式（categoryId、frequency）のタプル（タプル）の配列を取得しました。

 ArrayBuffer((314,3068), (132,2229), (128,1770), (270,1483), (139,1379), (107,1366), (177,1311), (226,1268), (103,1259), (127,1204))
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

さらに作業を進めるために、128番目のカテゴリーを選択することにしました。



データを準備します。グラフィックスがゴミで詰まらないように、必要な種類のオペレーティングシステムを除外します。



入り口で：

 val interestedBrowsers = List("Android", "OS X", "iOS", "Linux", "Windows") val osAov = dataAov.filter(x => interestedBrowsers.contains(x.osFamily)) //    .filter(_.categoryId == 128) //   .map(x => (x.osFamily, (x.aov, 1.0))) //      .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)) .map{ case(osFamily, (revenue, orders)) => (osFamily, revenue/orders) } .collect()
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

出力は、OS形式のタプルの配列、平均チェックです：

 Array((OS X,4859.827586206897), (Linux,3730.4347826086955), (iOS,3964.6153846153848), (Android,3670.8474576271187), (Windows,3261.030993042378))
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

視覚化が必要な場合は、HighChartsで行いましょう。



理論的には、 WispでサポートされているHighChartsグラフィックを使用できます。 すべてのグラフィックはインタラクティブです。



同じことを試してみましょう。ただし、Rを使用します。

Rクライアントを開始します。

 import org.ddahl.rscala._ import ru.retailrocket.ispark._ def connect() = RClient("R", false) @transient val r = connect()
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

チャート自体を作成します。



そのため、IPythonメモ帳で任意のRグラフィックを作成できます。

相互情報

グラフは依存関係があることを示していますが、メトリックはこの結論を確認しますか？ これを行うには多くの方法があります。 この場合、テーブル内の値の間で相互情報（ 相互情報 ）を使用します。 2つのランダム（離散）量の分布間の相互依存性を測定します。



離散分布の場合、次の式で計算されます。







しかし、より実用的なメトリックである最大情報係数 （MIC）に興味があります。これは、連続変数のトリックを計算するための計算です。 これは、このパラメーターの定義がどのように聞こえるかです。



D =（x、y）をランダム変数XおよびYの要素のnの順序付きペアのセットとします。この2次元空間はXおよびYグリッドで分割され、それぞれパーティションのXおよびYでxおよびyの値をグループ化します（ヒストグラムを思い出してください！）。







ここで、B（n）はグリッドのサイズ、I ∗（D、X、Y）はXとYのパーティションに関する相互情報です。分母は対数を示し、MICを間隔[0、1]の値に正規化するのに役立ちます。 MICは、間隔[0,1]で連続値を取ります。極端な値の場合、依存関係がある場合は1、そうでない場合は0です。 このトピックで他に読むことができるものは、記事の最後の参照リストにリストされています。



本では、 MIC （相互情報）は21世紀相関と呼ばれています。 そして、ここに理由があります！ 以下のグラフは、6つの依存関係を示しています（グラフC-H）。 それらについては、ピアソンとMICの相関が計算され、左側のグラフで対応する文字でマークされています。 ご覧のとおり、ピアソンの相関関係はほぼゼロですが、MICは依存関係を示しています（グラフF、G、E）。



ソース： people.cs.ubc.ca



以下の表は、ランダム、線形、キュービックなどの異なる依存関係で計算された多くのメトリックを示しています。 この表は、MICが非常に良好に動作し、非線形の依存関係を検出することを示しています。





別の興味深いグラフは、MICに対するノイズの影響を示しています。





私たちの場合、Aov変数が連続的であり、他のすべてが順序付けられていない値（ブラウザーのタイプなど）で離散している場合、MICの計算を処理しています。 MICを正しく計算するには、Aovをサンプリングする必要があります。 exploredata.netの既製のソリューションを使用します。 このソリューションには1つの問題があります。両方の変数が連続であり、Float値で表現されると見なされます。 したがって、離散量の値をFloatでエンコードし、これらの量の順序をランダムに変更することにより、コードを欺く必要があります。 これを行うには、ランダムな順序（100を実行）で多数の反復を実行する必要があり、その結果、最大のMIC値を取得します。

 import data.VarPairData import mine.core.MineParameters import analysis.Analysis import analysis.results.BriefResult import scala.util.Random //  ,    "" def encode(col: Array[String]): Array[Double] = { val ns = scala.util.Random.shuffle(1 to col.toSet.size) val encMap = col.toSet.zip(ns).toMap col.map{encMap(_).toDouble} } //   MIC def mic(x: Array[Double], y: Array[Double]) = { val data = new VarPairData(x.map(_.toFloat), y.map(_.toFloat)) val params = new MineParameters(0.6.toFloat, 15, 0, null) val res = Analysis.getResult(classOf[BriefResult], data, params) res.getMIC } //          def micMax(x: Array[Double], y: Array[Double], n: Int = 100) = (for{ i <- 1 to 100} yield mic(x, y)).max
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

さて、私たちは最終段階に近づいています。今度は計算自体を実行します。

 val aov = dataAov.filter(x => interestedBrowsers.contains(x.osFamily)) //    .filter(_.categoryId == 128) //   //osFamily var aovMic = aov.map(x => (x.osFamily, x.aov)).collect() println("osFamily MIC =" + micMax(encode(aovMic.map(_._1)), aovMic.map(_._2))) //orderId aovMic = aov.map(x => (x.orderId, x.aov)).collect() println("orderId MIC =" + micMax(encode(aovMic.map(_._1)), aovMic.map(_._2))) //cityId aovMic = aov.map(x => (x.cityId, x.aov)).collect() println("cityId MIC =" + micMax(encode(aovMic.map(_._1)), aovMic.map(_._2))) //uaName aovMic = aov.map(x => (x.uaName, x.aov)).collect() println("uaName MIC =" + mic(encode(aovMic.map(_._1)), aovMic.map(_._2))) //aov println("aov MIC =" + micMax(aovMic.map(_._2), aovMic.map(_._2))) //random println("random MIC =" + mic(aovMic.map(_ => math.random*100.0), aovMic.map(_._2)))
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

出力：

 osFamily MIC =0.06658 orderId MIC =0.10074 cityId MIC =0.07281 aov MIC =0.99999 uaName MIC =0.05297 random MIC =0.10599
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

実験のために、一様分布のランダム変数とAOV自体を追加しました。

ご覧のとおり、ほぼすべてのMICがランダムな値（ランダムMIC）を下回っていることが判明しました。これは「条件付き」決定しきい値と見なすことができます。 Aov MICはそれ自体との相関が1に等しいため、自然にほとんど統一されます。



興味深い質問が発生します。なぜグラフへの依存性があり、MICがゼロであるのでしょうか。 多くの仮説を思い付くことができますが、おそらくosファミリーのケースでは、すべてが非常に単純です-Windowsマシンの数は他をはるかに上回っています。

おわりに

Scalaがデータサイエンティストの人気を獲得することを願っています。 これは非常に便利です。標準のIPythonノートブックで作業して、Sparkのすべての機能を取得できるからです。 このコードはテラバイトのデータ配列で簡単に機能します。このため、クラスターのURIを指定して、ISparkの構成行を変更するだけです。



ところで、このエリアには空席があります。

• ジュニアアナリスト
• 研究アナリスト

便利なリンク：

MICの開発に基づいた科学記事 。

相互情報に関するKDnuggetsに関するメモ （ビデオがあります）。

PythonおよびMATLAB / OCTAVEのラッパーを使用してMICを計算するためのCライブラリ 。

MICが開発した科学論文の著者の サイト （ このサイトにはRのモジュールとJavaのライブラリがあります）。