Merge pull request #2 from xiaotaichai/lydia

Merge Normalized Revisions

Author: fbchow

blank_line_check.py

@@ -0,0 +1,20 @@
+from mrjob.job import MRJob
+
+
+class BlankLineCheck(MRJob):
+
+    def mapper(self, _, line):
+
+        line = line.strip()
+
+        if line:
+            yield 'Not blank', 1
+        else:
+            yield 'Blank', 1
+
+    def reducer(self, key, values):
+        yield key, sum(values)
+
+
+if __name__ == '__main__':
+    BlankLineCheck.run()

checking_format.py

@@ -0,0 +1,30 @@
+from mrjob.job import MRJob
+import gzip
+
+class CheckFormat(MRJob):
+
+    def mapper(self, _, line):
+
+        record = line.split('\x1e')
+        revision_info = record[0].split(' ')
+        record_length = len(record)
+        rev_info_length = len(revision_info)
+
+        if rev_info_length != 7:
+            if record_length != 13:
+                yield 'record length and revision info length is bad', 1
+            else:
+                yield 'revision info length bad, record length good', 1
+        else:
+            if record_length != 13:
+                yield 'record length bad, revision info length good', 1
+            else:
+                yield 'revision info length and record length good', 1
+
+
+    def reducer(self, key, values):
+        yield key, sum(values)
+
+
+if __name__ == '__main__':
+    CheckFormat.run()

create_normalized_revision_count_timeline.py

@@ -1,15 +1,18 @@
 from mrjob.job import MRJob
 from mrjob.protocol import TextValueProtocol
 
-class RandomSubsample(MRJob):
+class RevisionCountTimeline(MRJob):
 
     #OUTPUT_PROTOCOL = TextValueProtocol
 
     def mapper(self, _, line):
 
-        article_id = line.split('<<sep>>')[0]
-        article_name = line.split('<<sep>>')[2]
-        revision_date = line.split('<<sep>>')[3]
+        record = line.split('\x1e')
+        article_info = record[0].split(' ')
+
+        article_id = article_info[1]
+        article_name = article_info[3]
+        revision_date = article_info[4]
 
         yield [article_id, article_name], revision_date
 
@@ -19,9 +22,9 @@ def reducer(self, key, records):
             year, month = record.split('-')[0], record.split('-')[1]
             index = (int(year) - 2001)*12 + int(month) - 1
             monthly_revision_count[index] += 1
-        
+
         normalized = [float(i)/sum(monthly_revision_count) for i in monthly_revision_count]
         yield key, normalized
-        
+
 if __name__ == '__main__':
-    RandomSubsample.run()
+    RevisionCountTimeline.run()

create_normalized_revision_lengths_timeline.py

@@ -7,22 +7,30 @@ class RandomSubsample(MRJob):
 
     def mapper(self, _, line):
 
-        article_id = line.split('<<sep>>')[0]
-        article_name = line.split('<<sep>>')[2]
-        revision_date = line.split('<<sep>>')[3]
-        revision_length = line.split('<<sep>>')[-1]
+        #
+        # article_id = line.split('<<sep>>')[0]
+        # article_name = line.split('<<sep>>')[2]
+        # revision_date = line.split('<<sep>>')[3]
+        # revision_length = line.split('<<sep>>')[-1]
+        line = line.strip()
+        if line:
+            parts = line.split('<<sep>>')
+            article_id = parts[0]
+            article_name = parts[2]
+            revision_date = parts[3]
+            revision_length = parts[-1]
 
-        yield [article_id, article_name], [revision_date, revision_length]
+            yield [article_id, article_name], [revision_date, revision_length]
 
     def reducer(self, key, records):
         monthly_revision_count = [0]*96
         for record in records:
             year, month = record[0].split('-')[0], record[0].split('-')[1]
             index = (int(year) - 2001)*12 + int(month) - 1
             monthly_revision_count[index] += int(record[1])
-        
+
         normalized = [float(i)/sum(monthly_revision_count) for i in monthly_revision_count]
         yield key, normalized
-        
+
 if __name__ == '__main__':
     RandomSubsample.run()

creation_timelines_toCSV.py

@@ -0,0 +1,24 @@
+import os
+import re
+
+progress = 0
+outfile = open('./creation_timelines_5yrs.csv', 'w')
+# write header
+outfile.write('article_id,article_name,creation_datetime,num_revisions,all_revisions\n')
+
+line_pattern = '\["([0-9]+?)", "(.+?)"\]\s*\["([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2})", ([0-9]+?), \[(.+)\]\]'
+with open('./creation_timelines_5yrs_new.txt','r') as infile:
+    for line in infile:
+        parts = list(re.findall(line_pattern, line)[0])
+        # remove qutoes from the revision authour usernames
+        parts[4] = re.sub('"','',parts[4])
+
+        new_line = '{0},"{1}",{2},{3},"{4}"\n'.format(parts[0],parts[1],parts[2],parts[3],parts[4])
+        outfile.write(new_line)
+
+        progress += 1
+
+        if progress%10000 == 0:
+            print('{} lines processed so far'.format(progress))
+outfile.close()
+print('Done')

get_random_subsample.py

@@ -3,7 +3,7 @@
 import random
 
 # read in list of unique article id's
-with gzip.open('unique_all_articleids.gz','rt') as infile:
+with gzip.open('/Akamai_scratch/fanny_kevin_lydia_xiaotai/Wikipedia-Edits-Distributed-Computing/unique_all_articleids.gz','rt') as infile:
     all_ids = infile.readlines()
 
 # select n of those id's and strip the new lines

mrjob.conf

@@ -0,0 +1,7 @@
+runners:
+  local:
+    local_tmp_dir: /Akamai_scratch/
+    cleanup_on_failure: ALL
+  inline:
+    local_tmp_dir: /Akamai_scratch/
+    cleanup_on_failure: ALL

mrjob2.conf

@@ -0,0 +1,7 @@
+runners:
+  local:
+    local_tmp_dir: /Akamai/
+    cleanup_on_failure: ALL
+  inline:
+    local_tmp_dir: /Akamai/
+    cleanup_on_failure: ALL

revision_count_timeline_2.py

@@ -0,0 +1,76 @@
+from mrjob.job import MRJob
+from mrjob.step import MRStep
+# from mrjob.protocol import TextValueProtocol
+import datetime as dt
+
+class RevisionTimeline(MRJob):
+
+    # OUTPUT_PROTOCOL = TextValueProtocol
+
+    def mapperGroupRevisions(self, _, line):
+        record = line.split('\x1e')
+        article_info = record[0].split(' ')
+
+        article_id = article_info[1]
+        article_name = article_info[3]
+
+        revision_datetime_str = article_info[4]
+        user_name = article_info[5]
+        user_id = article_info[6]
+        revision_length = int(record[12].split(' ')[1])
+        minor_flag = int(record[11].split(' ')[1])
+
+        yield [article_id, article_name], [revision_datetime_str, revision_length, minor_flag, user_name, user_id]
+
+
+    def reducerCreateTimeline(self, key, revisions):
+        creation_datetime = dt.datetime.now()
+        revisions = list(revisions)
+
+        for r in revisions:
+            revision_datetime = dt.datetime.strptime(r[0],'%Y-%m-%dT%H:%M:%SZ')
+            if revision_datetime < creation_datetime:
+                creation_datetime = revision_datetime
+
+
+        # check if creation was before 01-01-03, so each article has at least a five year history
+
+        if creation_datetime < dt.datetime.strptime('2003-01-01','%Y-%m-%d'):
+            num_revisions = len(revisions)
+            normalized_revision_timeline = [[] for i in range(num_revisions)]
+
+            # creation_datetime = dt.datetime.strptime(values[1],'%Y-%m-%dT%H:%M:%SZ')
+            # revisions = values[0]
+            i = 0
+            for r in revisions:
+                revision_datetime = dt.datetime.strptime(r[0],'%Y-%m-%dT%H:%M:%SZ')
+                time_since_creation = revision_datetime - creation_datetime
+                normalized_revision_timeline[i] = [time_since_creation.days, time_since_creation.seconds, r[1], r[2],r[3],r[4]]
+                i += 1
+
+
+            yield key , [creation_datetime.strftime('%Y-%m-%d %H:%M:%S'), num_revisions, normalized_revision_timeline]
+        # yield key , [revisions, creation_datetime_str, num_revisions]
+
+    # def reducerCreateTimeline(self, key, values):
+    #
+    #     num_revisions = values[2]
+    #     normalized_revision_timeline = []*num_revisions
+    #
+    #     creation_datetime = dt.datetime.strptime(values[1],'%Y-%m-%dT%H:%M:%SZ')
+    #     revisions = values[0]
+    #     i = 0
+    #     for r in revisions:
+    #         revision_datetime = dt.datetime.strptime(r[0],'%Y-%m-%dT%H:%M:%SZ')
+    #         time_since_creation = revision_datetime - creation_datetime
+    #         normalized_revision_timeline[i] = [time_since_creation.days, time_since_creation.seconds, r[1], r[2]]
+    #         i += 1
+    #     yield key + (creation_datetime.strftime('%Y-%m-%d %H:%M:%S'),num_revisions), normalized_revision_timeline
+    def steps(self):
+        return [
+            MRStep(mapper=self.mapperGroupRevisions,
+                   reducer=self.reducerCreateTimeline)
+        ]
+
+if __name__ == '__main__':
+    RevisionTimeline.run()