我有表格中的数据
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
但是当我将它部署在集群上时。它没有说工作已被杀死..这个原因是未知的。
请帮忙。
谢谢。