機械学習の実践: Spark と Python を組み合わせるには?

Apache Sparkはビッグデータの処理や活用に最も広く使われているフレームワークの一つであり、Pythonはデータ分析や機械学習などの分野で最も広く使われているプログラミング言語の一つです。より強力な機械学習機能を手に入れたいなら、Spark と Python を一緒に使ってみてはいかがでしょうか?

海外では、Apache Spark 開発者の平均年収は 11 万ドルです。この業界で Spark が広く使用されていることは間違いありません。 Python は、豊富なライブラリを備えているため、ほとんどのデータ サイエンティストや分析の専門家にも使用されています。両者の統合はそれほど難しくありません。Spark は、Java に非常によく似た言語である Scala で開発されています。 Spark ビッグデータ処理用にプログラム コードを JVM バイトコードにコンパイルします。 Spark と Python を統合するために、Apache Spark コミュニティは PySpark をリリースしました。

Apache Spark は、Apache Software Foundation によって開発された、リアルタイム処理用のオープン ソース クラスター コンピューティング フレームワークです。 Spark は、暗黙的なデータ並列性とフォールト トレランスを備えたクラスター全体をプログラミングするためのインターフェイスを提供します。

他のフレームワークに比べて優れている Apache Spark の機能をいくつか紹介します。

• 速度: 従来の大規模データ処理フレームワークよりも 100 倍高速です。
• 強力なキャッシュ: シンプルなプログラミング レイヤーにより、強力なキャッシュ機能とディスク永続化機能が提供されます。
• デプロイメント: Mesos、Yarn、または Spark 独自のクラスター マネージャーを介してデプロイできます。
• リアルタイム: インメモリ コンピューティング、リアルタイム コンピューティング、低レイテンシ。
• ポリグロット: Scala、Java、Python、R でプログラミングできるため、これはこのフレームワークの最も重要な機能の 1 つです。

Spark は Scala で設計されており、Python よりも 10 倍高速ですが、Scala の速度の利点は、使用されるコアの数が少ない場合にのみ現れます。最近のほとんどの分析と処理には多数のコアが必要なので、Scala のパフォーマンス上の利点はそれほど大きくありません。

プログラマーにとって、Python は豊富な構文と標準ライブラリを備えているため、比較的簡単に学習できます。さらに、RDD は動的型付け言語であるため、複数のタイプのオブジェクトを保持できます。

Scala には SparkMLlib がありますが、機械学習や NLP を実装するためのライブラリやツールが十分にありません。さらに、Scala にはデータの視覚化機能が欠けています。

Python で Spark を設定する (PySpark)

まず、Sparkをダウンロードしてインストールする必要があります。Sparkファイルを抽出し、インストールして.bashrcファイルパスに追加したら、source .bashrcと入力する必要があります。

PySparkシェルを開くには、コマンド./bin/pysparkを入力します。

PySpark SparkContext とデータフロー

Python で Spark に接続するには、RD4s を使用し、ライブラリ Py4j を使用できます。 PySpark Shell は、Python API を Spark Core にリンクし、Spark コンテキストを初期化します。 SparkContext は Spark アプリケーションの中核です。

1. Spark Context は内部サービスをセットアップし、Spark 実行環境への接続を確立します。
2. ドライバー プログラム内の Spark Context オブジェクトは、すべての分散プロセスを調整し、リソースの割り当てを可能にします。
3. クラスター マネージャー エグゼキューターは、ロジックを持つ JVM プロセスです。
4. Spark Context オブジェクトは、アプリケーションをエグゼキューターに送信します。
5. Spark Context は各エグゼキュータでタスクを実行します。

PySpark KDD ユースケース

ここで、ユースケースを見てみましょう。データ ソースは KDD'99 Cup (国際知識発見およびデータ マイニング ツール コンペティション。中国には Zhihu など、同様のコンペティションのオープン データ セットがあります) です。元のデータセットが大きすぎるため、ここではデータセットの一部を取り出します。

RDD を作成する:

これで、このファイルを使用して RDD を作成できます。

フィルター

データセット内の通常の相互作用の数を数えたいとします。次のようにして raw_data RDD をフィルタリングできます。

カウント：

これで、新しい RDD に含まれる要素の数を数えることができます。

出力：

マッピング：

この場合、データファイルを CSV 形式のファイルとして読み取ります。これは、RDD 内の各要素にラムダ関数を適用することで実現できます。以下に示すように、ここでは map() および take() 変換を使用します。

出力：

スプリット：

ここで、RDD 内の各要素をキーと値のペアとして使用します。キーはタグ (通常の値など) で、値は CSV 形式ファイルの行を表す要素のリスト全体です。 これは、line.split() と map() を使用して次のように実行できます。

出力：

集める：

collect() アクションを使用して、RDD のすべての要素をメモリに保存します。したがって、大きな RDD を使用する場合は注意が必要です。

出力：

もちろん、これはこれまでのどの操作よりも時間がかかりました。 RDD の一部を持つ各 Spark ワーカー ノードは、その部分を取得してすべてをまとめるために調整する必要があります。

これまでのすべてを結び付ける最後の例として、通常のやり取りをすべてキーと値のペアとして収集します。

出力：