Amazon の EMR で mrjob を実行しようとしています。インライン ランナーを使用してローカルでジョブをテストしましたが、Amazon で実行すると失敗します。失敗を外部データ ファイルへの依存に絞り込みましたzip_codes.txt。ハードコーディングされた郵便番号データを使用してその依存関係なしで実行すると、問題なく動作します。
アップロード ファイル引数を使用して、必要なデータ ファイルを含めようとしました。S3 を見ると、ファイルはそこにありましたが、明らかに何か問題が発生しているため、ローカルでアクセスできません。
これが私のmrjob.confファイルです：
runners:
emr:
aws_access_key_id: FOOBARBAZQUX
aws_secret_access_key: IAMASECRETKEY
aws_region: us-east-1
ec2_key_pair: mapreduce
ec2_key_pair_file: $ENV/keys/mapreduce.pem
ssh_tunnel_to_job_tracker: true
ssh_tunnel_is_open: true
cleanup_on_failure: ALL
cmdenv:
TZ: America/Los_Angeles
これは私のMR_zip.pyファイルです。
from mrjob.job import MRJob
import mrjob
import csv
def distance(p1, p2):
# d = ...
return d
class MR_zip(MRJob):
OUTPUT_PROTOCOL = mrjob.protocol.JSONProtocol
zip_codes = {int(zip_code): (float(latitude), float(longitude)) for zip_code, latitude, longitude in csv.reader(open("zip_codes.txt", "r"))}
def mapper(self, _, line):
zip_code_1, poi = line.split(",")
zip_code_1 = int(zip_code_1)
lat1, lon1 = self.zip_codes[zip_code_1]
for zip_code_2, (lat2, lon2) in self.zip_codes.items():
d = distance((lat1, lon1), (lat2, lon2))
yield zip_code_2, (zip_code_1, poi, d)
def reducer(self, zip_code_1, ds):
result = {}
for zip_code_2, poi, d in ds:
if poi not in result:
result[poi] = (zip_code_2, d)
elif result[poi][1] > d:
result[poi] = (zip_code_2, d)
yield zip_code_1, result
if __name__ == '__main__':
MR_zip.run()
最後に、次のコマンドで実行します。
python MR_zip.py -r emr --conf mrjob.conf --file zip_codes.txt < poi.txt
zip_codes.txt は次のようになります。
...
62323,39.817702,-90.66923
62324,39.988988,-90.94976
62325,40.034398,-91.16278
62326,40.421857,-90.80333
...
poi.txt は次のようになります。
...
210,skate park
501,theatre
29001,theatre
8001,knitting club
20101,food bank
...