0

我有表格中的数据

  id,    movieid      , date,    time
 3710100, 13502, 2012-09-10, 12:39:38.000

现在基本上我想做的是这个..

我想知道，某部电影在上午 7 点到 11 点之间以 30 分钟的间隔观看了多少次

所以基本上..

电影之间看过多少次

  6 and 6:30
  6:30 and 7
   7 and 7:30
   ...
   10:30-11

所以我写了mapper和reducer来实现这个。

mapper.py

#!/usr/bin/env python

import sys

# input comes from STDIN (standard input)
for line in sys.stdin:
    # remove leading and trailing whitespace
    line = line.strip()
    # split the line into words
    line = line.split(",")
    #print line

    print '%s\t%s' % (line[1], line)

减速器.py

#!/usr/bin/env python

import sys
import datetime
from collections import defaultdict



def convert_str_to_date(time_str):
    try:
        timestamp =   datetime.datetime.strptime(time_str, '%Y-%m-%d:%H:%M:%S.000')  #00:23:51.000


        return timestamp

    except Exception,inst:

        pass

def is_between(time, time1,time2):
    return True if time1 <= time < time2 else False


def increment_dict(data_dict, se10,date_time):
    start_time = datetime.datetime(date_time.year,date_time.month,date_time.day, 07,00,00)
    times = [start_time]
    for i in range(8):
        start_time += datetime.timedelta(minutes = 30 )
        times.append(start_time)
    for i in range(len(times) -1 ):
        if is_between(date_time, times[i], times[i+1]):
            data_dict[se10][i] += 1






keys = [0,1,2,3,4,5,6,7]



data_dict = defaultdict(dict)


# input comes from STDIN
def initialize_entry(se10):
    for key in keys:
        data_dict[se10][key] = 0

for line in sys.stdin:
    # remove leading and trailing whitespace
    line = line.strip()


    # parse the input we got from mapper.py

    se10, orig_data = line.split('\t')
    initialize_entry(se10)
    parse_line = orig_data.split(",")

    datestr = parse_line[2].replace(" ","").replace("'","")
    timestr = parse_line[3].replace(" ","").replace("'","")

    date_time = datestr + ":" + timestr

    time_stamp = convert_str_to_date(date_time)

    increment_dict(data_dict, se10,time_stamp)


for key, secondary_key in data_dict.items():
    for skey, freq in secondary_key.items():
        print key,"," ,skey,",",freq

如果我这样做，上面的代码运行得很好

   cat input.txt | python mapper.py | sort | python reducer.py

但是当我将它部署在集群上时。它没有说工作已被杀死..这个原因是未知的。

请帮忙。

谢谢。

4

2 回答 2

0

好的，我想通了这件事..

主要问题是我的工作本地机器是基于 Windows 的.. 而集群是基于 linux 的..

所以我不得不将用dos编写的文件转换为unix格式..

于 2012-11-21T01:06:37.217 回答
0

通读 JobHistory 中的日志通常是个好主意，如https://stackoverflow.com/a/24509826/1237813中所述。它应该为您提供更多详细信息，为什么作业失败。

关于行尾，Hadoop Streaming 默认使用的分割行的类是TextInputFormat。它曾经与 Windows 换行符中断，但自 2006 年以来它应该可以正常工作。

这使您的映射器和减速器脚本成为问题的可能来源。Python 3 使用了一种叫做通用换行符的东西，它应该对 Unix 和 Windows 换行符都开箱即用。在 Python 2.7 中，您需要显式打开它。

在 Linux 和 Mac OS X 上，您可以像这样启用通用换行符重新打开标准输入sys.stdin = open('/dev/stdin', 'U')。我手头没有可以尝试的 Windows 计算机，但以下内容应该适用于所有三个系统：

import os
import sys

# reopen sys.stdin
os.fdopen(sys.stdin.fileno(), 'U')

for line in sys.stdin:
    …
于 2014-07-03T13:21:16.220 回答