我有表格里的数据

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

但当我把它部署到集群上时。不能说工作已经结束了。。这个原因还不清楚。
请帮忙。
谢谢。