-
Notifications
You must be signed in to change notification settings - Fork 0
/
mrcc.py
43 lines (39 loc) · 1.32 KB
/
mrcc.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import gzip
import os.path as Path
#
import boto
import warc
#
from boto.s3.key import Key
from gzipstream import GzipStreamFile
from mrjob.job import MRJob
class CCJob(MRJob):
def configure_options(self):
super(CCJob, self).configure_options()
self.pass_through_option('--runner')
self.pass_through_option('-r')
def process_record(self, record):
"""
Override process_record with your mapper
"""
raise NotImplementedError('Process record needs to be customized')
def mapper(self, _, line):
f = None
## If we're on EC2 or running on a Hadoop cluster, pull files via S3
if self.options.runner in ['emr', 'hadoop']:
# Connect to Amazon S3 using anonymous credentials
conn = boto.connect_s3(anon=True)
pds = conn.get_bucket('commoncrawl')
# Start a connection to one of the WARC files
k = Key(pds, line)
f = warc.WARCFile(fileobj=GzipStreamFile(k))
## If we're local, use files on the local file system
else:
line = Path.join(Path.abspath(Path.dirname(__file__)), line)
print('Loading local file {}'.format(line))
f = warc.WARCFile(fileobj=gzip.open(line))
###
for i, record in enumerate(f):
for key, value in self.process_record(record):
yield key, value
self.increment_counter('commoncrawl', 'processed_records', 1)