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.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)

reducer.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 に答える