使用 python 代码在 mapreduce 中没有得到我预期的输出

Question

运行此代码用于获取 Hadoop 中的概率集群我在 CSV 文件中的数据大约 10k+。我正在使用 Google DataProc Cluster 来运行这段代码。请告诉我如何获得预期的输出。最后一件事可能存在逻辑问题或某些功能问题。

#!/usr/bin/env python3
"""mapper.py"""
import sys

# Get input lines from stdin
for line in sys.stdin:
    # Remove spaces from beginning and end of the line
    line = line.strip()

    # Split it into tokens
    #tokens = line.split()

    #Get probability_mass values
    for probability_mass in line:
        print("None\t{}".format(probability_mass))

#!/usr/bin/env python3
"""reducer.py"""
import sys
from collections import defaultdict


counts = defaultdict(int)

# Get input from stdin
for line in sys.stdin:
    #Remove spaces from beginning and end of the line
    line = line.strip()

    # skip empty lines
    if not line:
        continue  

    # parse the input from mapper.py
    k,v = line.split('\t', 1)
    counts[v] += 1

total = sum(counts.values())
probability_mass = {k:v/total for k,v in counts.items()}
print(probability_mass)

我的 CSV 文件如下所示。

probability_mass
10
10
60
10
30

Expected output Probability of each number

{10: 0.6, 60: 0.2, 30: 0.2}

but result still show like this 
{1:0} {0:0} {3:0} {6:0} {1:0} {6:0}

我会在 nano 中保存这个命令，然后运行这个。

yarn jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar \
-D mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator \
-D mapred.text.key.comparator.options=-n \
-files mapper.py,reducer.py \
-mapper "python mapper.py" \
-reducer "python reducer.py" \
-input /tmp/data.csv \
-output /tmp/output

Answer 1

您将行拆分为单个字符，这可以解释为什么您得到 1、3、6、0 等作为映射键。

不要循环，只打印值所在的行；你的映射器不需要超过这个

import sys
for line in sys.stdin:
    print("None\t{}".format(line.strip()))

然后，在 reducer 中，您将一个 int 除以一个更大的 int，结果向下舍入到最接近的 int，即 0。

您可以通过将字典更改为存储浮点数来解决此问题

counts = defaultdict(float)

或将总和设为浮点数

total = float(sum(counts.values()))

如前所述，这不是 Hadoop 问题，因为您可以在本地测试和调试它

cat data.txt | python mapper.py | sort -n | python reducer.py

使用 python 代码在 mapreduce 中没有得到我预期的输出

Not getting my expected output in mapreduce using python code

python

python-3.x

hadoop

mapreduce

hadoop-streaming