mapreduceとpysparkを使って、ある年のある日の頻度を見つける方法

Question

私は、各行に日付を表​​す文字列を含むテキストファイル（61Gb）を持っています。 Thu Dec 16 18:53:32 +0000 2010

ファイルを1つのコアに反復するには時間がかかりすぎるため、PysparkとMapreduceテクノロジを使用して特定の年。私が何を考えて

は良いスタートです：

import dateutil.parser 
text_file = sc.textFile('dates.txt') 
date_freqs = text_file.map(lambda line: dateutil.parser.parse(line)) \ 
     .map(lambda date: date + 1) \ 
     .reduceByKey(lambda a, b: a + b) 

残念ながら、私は特定の年にフィルタリングし、キーによって軽減する方法を理解することはできません。キーはその日です。

出力例：本の線に沿って

木12月16日26543

木12月17日345

等

Answer 1

としてanother answerで言及されているように、dateutil.parser.parseはdatetime objectを返します。year、month、およびday属性：

>>> dt = dateutil.parser.parse('Thu Dec 16 18:53:32 +0000 2010') 
>>> dt.year 
2010 
>>> dt.month 
12 
>>> dt.day 
16 

このRDDを皮切り：

>>> rdd = sc.parallelize([ 
...  'Thu Oct 21 5:12:38 +0000 2010', 
...  'Thu Oct 21 4:12:38 +0000 2010', 
...  'Wed Sep 22 15:46:40 +0000 2010', 
...  'Sun Sep 4 22:28:48 +0000 2011', 
...  'Sun Sep 4 21:28:48 +0000 2011']) 

は、ここでは、すべての年 - 月 - 日の組み合わせのためのカウントを取得することができます方法は次のとおりです。

>>> from operator import attrgetter 
>>> counts = rdd.map(dateutil.parser.parse).map(
...  attrgetter('year', 'month', 'day')).countByValue() 
>>> counts 
defaultdict(<type 'int'>, {(2010, 9, 22): 1, (2010, 10, 21): 2, (2011, 9, 4): 2}) 

あなたが望む出力を得るには：

>>> for k, v in counts.iteritems(): 
...  print datetime.datetime(*k).strftime('%a %b %y'), v 
... 
Wed Sep 10 1 
Thu Oct 10 2 
Sun Sep 11 2 

あなたが唯一の特定の年のためのカウントをしたい場合、あなたは、カウントを行う前に、RDDをフィルタリングすることができます。

>>> counts = rdd.map(dateutil.parser.parse).map(
... attrgetter('year', 'month', 'day')).filter(
... lambda (y, m, d): y == 2010).countByValue() 
>>> counts 
defaultdict(<type 'int'>, {(2010, 9, 22): 1, (2010, 10, 21): 2}) 

Answer 2

何かが良いスタートかもしれない：

import dateutil.parser 
text_file = sc.textFile('dates.txt') 
date_freqs = text_file.map(lambda line: dateutil.parser.parse(line)) 
    .keyBy((_.year, _.month, _.day)) // somehow get the year, month, day to key by 
    .countByKey() 

Answer 3

私はdateutilは、Pythonで、標準ではないことを追加する必要があります。あなたのクラスタにsudoがない場合、これは問題を引き起こす可能性があります。解決策として、私はdatetime型を使用して提案したい：

import datetime 
def parse_line(d): 
    f = "%a %b %d %X %Y" 
    date_list = d.split() 
    date = date_list[:4] 
    date.append(date_list[5]) 
    date = ' '.join(date) 
    return datetime.datetime.strptime(date, f) 

counts = rdd.map(parse_line)\ 
    .map(attrgetter('year', 'month', 'day'))\ 
    .filter(lambda (y, m, d): y == 2015)\ 
    .countByValue() 

私はより良いソリューションに興味を持って使用して：寄木張りを、行/列など