PySpark-MLlib

Apache Sparkは、 MLlib と呼ばれる機械学習APIを提供します。 PySparkには、Pythonでこの機械学習APIもあります。以下に記載されているさまざまな種類のアルゴリズムをサポートしています-

mllib.classification - spark.mllib パッケージは、バイナリ分類、マルチクラス分類、および回帰分析のためのさまざまなメソッドをサポートしています。分類で最も人気のあるアルゴリズムのいくつかは、 Random Forest、Naive Bayes、Decision Tree などです。
mllib.clustering -クラスタリングは教師なしの学習問題であり、類似性の概念に基づいてエンティティのサブセットを互いにグループ化することを目的としています。
mllib.fpm -頻繁なパターンマッチングは、大規模なデータセットを分析するための最初のステップに通常含まれる、頻繁なアイテム、アイテムセット、サブシーケンス、またはその他の下位構造をマイニングすることです。これは何年もの間、データマイニングの活発な研究トピックです。
mllib.linalg -線形代数用のMLlibユーティリティ。
mllib.recommendation -レコメンダーシステムでは、協調フィルタリングが一般的に使用されます。これらの手法は、ユーザーアイテムの関連付けマトリックスの不足しているエントリを埋めることを目的としています。
spark.mllib -現在、モデルベースの協調フィルタリングをサポートしています。このフィルタリングでは、ユーザーと製品は、不足しているエントリの予測に使用できる潜在的な要因の小さなセットで記述されます。 spark.mllibは、交互最小二乗（ALS）アルゴリズムを使用してこれらの潜在的要因を学習します。
mllib.regression -線形回帰は回帰アルゴリズムのファミリーに属します。回帰の目的は、変数間の関係と依存関係を見つけることです。線形回帰モデルとモデルの要約を操作するためのインターフェイスは、ロジスティック回帰の場合に似ています。

mllibパッケージの一部として、他のアルゴリズム、クラス、および関数もあります。今のところ、 pyspark.mllib のデモを理解しましょう。

次の例は、ALSアルゴリズムを使用して推奨モデルを構築し、トレーニングデータで評価する協調フィルタリングの例です。

使用されるデータセット-test.data

1,1,5.0
1,2,1.0
1,3,5.0
1,4,1.0
2,1,5.0
2,2,1.0
2,3,5.0
2,4,1.0
3,1,1.0
3,2,5.0
3,3,1.0
3,4,5.0
4,1,1.0
4,2,5.0
4,3,1.0
4,4,5.0

--------------------------------------recommend.py----------------------------------------
from __future__ import print_function
from pyspark import SparkContext
from pyspark.mllib.recommendation import ALS, MatrixFactorizationModel, Rating
if __name__ == "__main__":
   sc = SparkContext(appName="Pspark mllib Example")
   data = sc.textFile("test.data")
   ratings = data.map(lambda l: l.split(','))\
      .map(lambda l: Rating(int(l[0]), int(l[1]), float(l[2])))

   # Build the recommendation model using Alternating Least Squares
   rank = 10
   numIterations = 10
   model = ALS.train(ratings, rank, numIterations)

   # Evaluate the model on training data
   testdata = ratings.map(lambda p: (p[0], p[1]))
   predictions = model.predictAll(testdata).map(lambda r: ((r[0], r[1]), r[2]))
   ratesAndPreds = ratings.map(lambda r: ((r[0], r[1]), r[2])).join(predictions)
   MSE = ratesAndPreds.map(lambda r: (r[1][0] - r[1][1])**2).mean()
   print("Mean Squared Error = " + str(MSE))

   # Save and load model
   model.save(sc, "target/tmp/myCollaborativeFilter")
   sameModel = MatrixFactorizationModel.load(sc, "target/tmp/myCollaborativeFilter")
--------------------------------------recommend.py----------------------------------------

コマンド-コマンドは次のようになります-

$SPARK_HOME/bin/spark-submit recommend.py

出力-上記のコマンドの出力は-

Mean Squared Error = 1.20536041839e-05

Pyspark-mllib

PySpark-MLlib