Python + Hadoop ストリーミングによる分散プログラミング (1) -- 原理、サンプルプログラム、ローカルデバッグの紹介

MapReduce と HDFS の概要
Hadoopとは何ですか?

Google は、自社のビジネス ニーズのためにプログラミング モデル MapReduce と分散ファイル システム Google File System を提案し、関連論文を出版しました (Google Research Web サイト: GFS、MapReduce で入手可能)。 Doug Cutting と Mike Cafarella は、検索エンジン Nutch の開発時に、これら 2 つの論文、つまり同じ名前の MapReduce と HDFS (合わせて Hadoop) を独自に実装しました。

MapReduce のデータの流れは以下の通りです。元のデータは Mapper で処理され、分割、ソートされ、Reducer に到達し、最終的な結果が出力されます。

Hadoop からの画像: 決定版ガイド

Hadoop ストリーミングの原則
Hadoop 自体は Java で開発されており、プログラムも Java で記述する必要がありますが、Hadoop ストリーミングを使用すると、任意の言語を使用してプログラムを作成し、Hadoop を実行できます。

Hadoop ストリーミング関連のソース コードは、Hadoop の Github リポジトリで参照できます。簡単に言うと、他の言語で書かれたマッパーとリデューサーは、事前に作成された Java プログラム (Hadoop に付属の *-streaming.jar) にパラメーターを通じて渡されます。この Java プログラムは、MR ジョブの作成とオープンを担当します。別のプロセスでマッパーを実行し、stdin 経由でマッパーに入力を渡し、マッパーによって処理されたデータを stdout から Hadoop に転送し、分割して並べ替えます。次に、別のプロセスを開いてリデューサーを実行し、同様に stdin 経由で最終結果を取得します。 /stdout 。したがって、他の言語で書かれたプログラムでは stdin を介してデータを受信し、処理されたデータを stdout に出力するだけで済みます。Hadoop ストリーミングを使用すると、面倒な中間ステップを解決し、この Java ラッパーを介して分散プログラムを実行できます。

Hadoop からの画像: 決定版ガイド

原則として、stdio を処理できる任意の言語を使用してマッパーとリデューサーを作成できます。また、Linux 上のプログラム (awk、grep、cat など) または特定の Java クラスとしてマッパーまたはリデューサーを指定することもできます。形式。したがって、マッパーとリデューサーは同じ種類のプログラムである必要はありません。

Hadoop ストリーミングの長所と短所

メリット

MapReduce プログラムは好みの言語で作成できます (つまり、Java XD を作成する必要はありません)
Java MR プログラムを作成する場合のように、多くのライブラリをインポートしたり、コード内で多くの設定を行う必要はなく、多くのことが stdio に抽象化され、コードの量が大幅に削減されます
。 ライブラリの依存関係がないためデバッグが便利で、Hadoop を使用せずにローカルでパイプライン シミュレーションのデバッグを使用できます

デメリット

MapReduce フレームワークは、コード内で API を使用できる Java プログラムとは異なり、コマンド ライン パラメーターを通じてのみ制御できます。制御は比較的弱く、一部のことは私たちの手の届かないものです。
間に処理層があるため、効率が遅くなります
したがって、Hadoop ストリーミングは、Python でわずか 100 ～ 200 行のスクリプトを作成するなど、いくつかの単純なタスクを実行するのに適しています。プロジェクトがより複雑である場合、またはより詳細な最適化が必要な場合、ストリーミングの使用時にいくつかの制約が発生しやすくなります。

Python で簡単な Hadoop ストリーミング プログラムを作成します

ここでは 2 つの例を示します:

Michael Noll の単語カウント プログラム
Hadoop: 決定版ガイドのルーチン
Python を使用して Hadoop ストリーミング プログラムを作成する場合は、いくつかの注意点があります。

イテレータが使用可能な場合は、大量の標準入力入力をメモリに保存しないようにイテレータを使用するようにしてください。そうしないと、パフォーマンスが大幅に低下します。

ストリーミングでは、キーと値を分割して渡すことはできません。渡されるのは単なる文字列です。コード内で手動で split() を呼び出す必要があります。

標準入力から取得したデータの各行の末尾には n があるようですが、安全を期すために、通常は rstrip() を使用してそれを削除する必要があります。

キーと値のペアを 1 つずつ処理するのではなく、K-V リストを取得したい場合は、itemgetter で groupby を使用して、同じキーを持つ k-v ペアのグループを形成できます。Text タイプのキーと Reduce を取得できます。 Java の値としての反復可能の効果に似ています。 itemgetter はラムダ式より効率的であるため、要件がそれほど複雑でない場合は、可能な限り itemgetter を使用することをお勧めします。

Hadoop ストリーミング プログラムを作成するときの私の基本テンプレートは

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Some description here...
"""

import sys
from operator import itemgetter
from itertools import groupby

def read_input(file):
 """Read input and split."""
 for line in file:
  yield line.rstrip().split('\t')

def main():
 data = read_input(sys.stdin)
 for key, kviter in groupby(data, itemgetter(0)):
  # some code here..

if __name__ == "__main__":
 main()

デフォルトとは異なる入出力形式を制御できる場合は、主に read_input() で調整されます。

ローカルデバッグ

Hadoop ストリーミングの Python プログラムのローカル デバッグの基本モードは次のとおりです:

$ cat <input path> | python <path to mapper script> | sort -t $'\t' -k1,1 | python <path to reducer script> > <output path>

または、冗長な cat を使用したくない場合は、 を使用することもできます。

$ python <path to mapper script> < <input path> | sort -t $'\t' -k1,1 | python <path to reducer script> > <output path>

ここで注意すべき点がいくつかあります:

Hadoop はデフォルトでキーと値をタブで分割し、最初に分割した部分をキーとしてキーごとにソートするため、ここでは

を使用します

sort -t $'t' -k1,1
シミュレーションするために。他に必要な場合は、実行のために Hadoop ストリーミングに引き渡すときにコマンド ライン パラメーターを調整できます。また、ローカル デバッグ中に、主に並べ替えパラメーターを調整することで、対応する調整を行うこともできます。したがって、ローカル デバッグに習熟するには、まず sort コマンドの使用法をマスターすることをお勧めします。

如果你在python脚本里加上了shebang，并且为它们添加了执行权限，也可以用类似于

./mapper.py

来代替

python mapper.py