Daniel Kinzler has submitted this change and it was merged.
Change subject: Final wikiparser mapper and reducer modules using classes.
Added more docs. Reduced and polished code.
......................................................................
Final wikiparser mapper and reducer modules using classes. Added more docs.
Reduced and polished code.
Change-Id: I33d8b0aa1927a07a08b405905f6e65a14632b116
---
D wikiparser/disco/chunk.py
D wikiparser/disco/wikiparser.py
A wikiparser/mapper.py
A wikiparser/reducer.py
A wikiparser/runhadoop.sh
D wikiparser/wikiparser.py
D wikiparser/wikiparser_db.py
D wikiparser/wikiparser_r.py
8 files changed, 422 insertions(+), 347 deletions(-)
Approvals:
Daniel Kinzler: Verified; Looks good to me, approved
diff --git a/wikiparser/disco/chunk.py b/wikiparser/disco/chunk.py
deleted file mode 100644
index f979766..0000000
--- a/wikiparser/disco/chunk.py
+++ /dev/null
@@ -1,41 +0,0 @@
-import sys, os
-from disco.ddfs import DDFS
-
-
-if(len(sys.argv) < 2):
- print "USAGE: python", sys.argv[0], "<ddfs tag:name> [<uncompressed
wikidata dump file path>]"
- print "You may pipe the dump file contents using lbzcat or bzcat into
stdin like:"
- print "lbzcat wikidatawiki-yyyymmdd-pages-meta-current.xml.bz2 | python
<ddfs tag:name>\n"
- sys.exit()
-
-tag = (sys.argv[1])
-infile = sys.stdin if len(sys.argv) < 3 else sys.argv[2]
-
-getChunk = lambda x : sys.argv[1] + ".chunk" + str(x)
-fcount = 1
-count = 1
-outfile = open(getChunk(fcount), "w")
-
-for i in infile:
- outfile.write(i)
- if '</page>' in i:
- count += 1
- printProgress = True;
-
- if count % 100000 == 0 and printProgress == True:
- sys.stdout.write("\r%s" % "Finished " + str(count) + " pages.")
- sys.stdout.flush()
- printProgress = True if count % 1000000 == 0 else False
-
- if count % 1000000 == 0 and printProgress == True:
- outfile.close()
- DDFS(master="disco://localhost").push(tag,
[os.path.abspath(getChunk(fcount))])
- sys.stdout.write("\nChunk " + str(fcount) + " pushed to tag %s on
DDFS.\n" % tag)
- sys.stdout.flush()
- os.remove(getChunk(fcount))
- fcount += 1
- outfile = open(getChunk(fcount), "w")
- printProgress = False
- break
-
-if len(sys.argv) > 2: infile.close()
diff --git a/wikiparser/disco/wikiparser.py b/wikiparser/disco/wikiparser.py
deleted file mode 100644
index 9938f70..0000000
--- a/wikiparser/disco/wikiparser.py
+++ /dev/null
@@ -1,147 +0,0 @@
-from disco.core import Job, result_iterator
-import sys
-
-class WikiParser(Job):
- partitions = 4
-
- def map(self, row, params):
- results = self._parsePage("<page>" + row[0] + "</page>")
- if results != None:
- for j in results:
- if j != None:
- yield j[0], j[1]
-
- @staticmethod
- def regex_reader(item_re_str, fd, size, fname, output_tail=False,
read_buffer_size=8192):
- """
- Modified version of disco.worker.classic.func.re_reader - does not
throw DataError
- A map reader that uses an arbitrary regular expression to parse the
input
- stream.
-
- :param item_re_str: regular expression for matching input items
-
- The reader works as follows:
-
- 1. X bytes is read from *fd* and appended to an internal buffer *buf*.
- 2. ``m = regexp.match(buf)`` is executed.
- 3. If *buf* produces a match, ``m.groups()`` is yielded, which
contains an
- input entry for the map function. Step 2. is executed for the remaining
- part of *buf*. If no match is made, go to step 1.
- 4. If *fd* is exhausted before *size* bytes have been read,
- and *size* tests ``True``, instead of raising a
:class:`disco.error.DataError`,
- it prints something about the error.
- 5. When *fd* is exhausted but *buf* contains unmatched bytes, two
modes are
- available: If ``output_tail=True``, the remaining *buf* is yielded as
is.
- Otherwise, a message is sent that warns about trailing bytes.
- The remaining *buf* is discarded.
-
- Note that :func:`re_reader` fails if the input streams contains
unmatched
- bytes between matched entries.
- Make sure that your *item_re_str* is constructed so that it covers all
- bytes in the input stream.
-
- :func:`re_reader` provides an easy way to construct parsers for textual
- input streams.
- For instance, the following reader produces full HTML
- documents as input entries::
-
- def html_reader(fd, size, fname):
- for x in re_reader("<HTML>(.*?)</HTML>", fd, size, fname):
- yield x[0]
-
- """
- item_re = re.compile(item_re_str)
- buf = b""
- tot = 0
- while True:
- if size:
- r = fd.read(min(read_buffer_size, size - tot))
- else:
- r = fd.read(read_buffer_size)
- tot += len(r)
- buf += r
-
- m = item_re.match(buf)
- while m:
- yield m.groups()
- buf = buf[m.end():]
- m = item_re.match(buf)
-
- if not len(r) or (size!=None and tot >= size):
- if size != None and tot < size:
- print("Truncated input: Expected {0} bytes, got
{1}".format(size, tot), fname)
- if len(buf):
- if output_tail:
- yield [buf]
- else:
- print("Couldn't match the last {0} bytes in {1}. "
- "Some bytes may be missing from
input.".format(len(buf), fname))
- break
-
- @staticmethod
- def map_reader(fd, size, url, params):
- from wikiparser import WikiParser
- count = 0
- line = fd.readline()
- while "<page>" not in line:
- line = fd.readline()
- count += 1
-
- fd.seek(0);
-
- while count > 0:
- fd.readline()
- count -= 1
-
- reader = WikiParser.regex_reader("\s\s<page>([\s\S]*?)</page>\\n", fd,
size, url);
- for row in reader:
- yield row
-
- def reduce2(self, rows_iter):
- for i in rows_iter:
- yield i
-
- def _parsePage(self, page):
- from lxml import etree
- from StringIO import StringIO
- import json
-
- tree = etree.parse(StringIO(page))
- page = {child.tag:child.text for child in tree.iter()}
- title = page['title'][1:]
- try:
- if page['ns'] == '0':
- text = json.loads(page['text'])
- statement = None
- if 'claims' in text:
- for a in text['claims']:
- statement = {i:j for i, j in self._pairwise(a['m'])}
- if statement != None:
- try:
- toyield1 = (title, str(statement['value']))
- value =
str(statement['wikibase-entityid']['numeric-id']) if 'wikibase-entityid' in
statement else statement['string']
- toyield2 = (title, str(statement['value']) +
"----" + value)
- except KeyError:
- toyield1 = toyield2 = None
- yield toyield1
- yield toyield2
- except KeyError:
- pass
-
- def _pairwise(self, iterable):
- from itertools import izip
- a = iter(iterable)
- return izip(a, a)
-
-if __name__ == '__main__':
- from wikiparser import WikiParser
-
- if(len(sys.argv) < 2):
- print "USAGE: python", sys.argv[0], "<ddfs tag:name> [<output file
path>]"
- print "You may omit the output file; it's stdout by default.\n"
- sys.exit()
-
- job = WikiParser().run(input=[sys.argv[1]])
- outf = sys.stdout if len(sys.argv) < 3 else open(sys.argv[2], "w")
- for a, b in result_iterator(job.wait(show=True)):
- outf.write(a.encode('utf-8') + "," + b.encode('utf-8') + "\n")
diff --git a/wikiparser/mapper.py b/wikiparser/mapper.py
new file mode 100644
index 0000000..84f9e50
--- /dev/null
+++ b/wikiparser/mapper.py
@@ -0,0 +1,245 @@
+#!/usr/bin/python
+
+"""This is the mapper module for running the wikiparser Hadoop scripts. See
main() for usage info.
+The org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat
class in wikiparser-0.1.jar
+is an InputFormat class that is used by Hadoop to split the input XML data
dump into chunks of <page>...</page>.
+In simple terms, if there are N mappers and M chunks, Hadoop divides them into
the N mappers.
+
+This module contains an abstract class with a public run() method. It reads
the chunks of <page>...</page> from
+stdin and calls _getClaimsAndTitle on a single chunk, which in turn extracts
the claims from a page (if it's a wikibase Item)
+and calls _generateKeyValues on the list of claims. _generateKeyValues is
implemented in the different subclasses to yield
+different kinds of CSV rows for the different datasets.
+"""
+
+from lxml import etree
+from itertools import izip
+from StringIO import StringIO
+import json
+import sys
+import traceback
+
+def main():
+ """The following are 6 types of sample commands. Note that the -input and
-output parameters in the Hadoop command
+ are paths on HDFS and not on the local filesystem. Replace /path/to with
the actual paths of the files.
+ wikiparser-0.1.jar, mapper.py and reducer.py have to available in a
location where Hadoop has enough permissions to access.
+
+ 1. For generating the item-property pairs using Hadoop:
+ (Dataset used to suggest global claim properties)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/global-ip-pairs \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py global-ip-pairs' \
+ -reducer '/bin/cat'
+
+ 2. For generating the claim guid-property pairs for source refs using
Hadoop:
+ (Dataset used to suggest source ref properties)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/ref-cp-pairs \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py ref-cp-pairs' \
+ -reducer '/bin/cat'
+
+ 3. For counting frequencies of properties used in source refs:
+ (Dataset used to suggest source ref properties for empty source refs)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/ref-props \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py ref-props' \
+ -file /path/to/reducer.py -reducer '/path/to/reducer.py ref-props'
+
+ 4. For counting global property (used in all claims) frequencies:
+ (Dataset used to suggest global claim properties for empty items)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/global-props \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py global-props' \
+ -file /path/to/reducer.py -reducer '/path/to/reducer.py global-props'
+
+ 5. For counting frequency of properties used in qualifiers, to find the
top N popular qualifier properties for each property:
+ (Dataset used to suggest qualifiers, given a property)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/qual-props \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py qual-props' \
+ -file /path/to/reducer.py -reducer '/path/to/reducer.py qual-props'
+
+ 6. For counting frequency of values used with properties, to find the top
N popular values for each property:
+ (Dataset used to suggest values, given a property)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/prop-values \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py prop-values' \
+ -file /path/to/reducer.py -reducer '/path/to/reducer.py prop-values'
+ """
+
+ mapperClass = {
+ 'global-ip-pairs': GlobalIPPairGeneratorMapper,
+ 'ref-cp-pairs': RefClaimPropPairGeneratorMapper,
+ 'ref-props': CountRefPropsMapper,
+ 'global-props': CountGlobalPropsMapper,
+ 'qual-props': CountQualiferPropsMapper,
+ 'prop-values': CountPropertyValuesMapper
+ }
+
+ try:
+ mapper = mapperClass[sys.argv[1]]()
+ except (KeyError, IndexError):
+ print "Unrecognized/missing mapper arguments."
+ print "Please enter any of global-ip-pairs, ref-cp-pairs, ref-props,
global-props, qual-props, prop-values."
+ return
+
+ mapper.run()
+
+class AbstractWikiParserMapper:
+ """Abstract class that should be inherited by all mapper classes"""
+
+ def run(self):
+ """Main driver method that iterates through the pages and sends them
to the generateKeyValues method."""
+ while True:
+ page = ''
+ i = sys.stdin.readline()
+ if not i: break
+ if '<page>' in i:
+ i = sys.stdin.readline()
+ while '</page>' not in i:
+ page += i
+ i = sys.stdin.readline()
+ page = '<page>' + page + '</page>'
+ (claims, title) = self._getClaimsAndTitle(page)
+ if claims != None: self._generateKeyValues(claims, title)
+ sys.stdout.flush()
+
+ def _getClaimsAndTitle(self, page):
+ """Parses a page and its JSON text to return the list of claims.
"""
+ tree = etree.parse(StringIO(page))
+ page = {child.tag:child.text for child in tree.iter()}
+ if page['ns'] == '0':
+ title = page['title'][1:]
+ text = json.loads(page['text'])
+ if 'claims' in text:
+ return text['claims'], title
+ else:
+ return None, None
+
+ def _generateKeyValues(self, claims, title):
+ """Iterates through a list of claims, performs some operations and
writes (key, value) pairs
+ to stdout to be fed into the reducer script."""
+ raise NotImplementedError("Please implement this method in a
subclass.")
+
+ def _pairwise(self, iterable):
+ a = iter(iterable)
+ return izip(a, a)
+
+class GlobalIPPairGeneratorMapper(AbstractWikiParserMapper):
+ """The mapper class for generating the item-property pairs and
item-(property-value) pairs using Hadoop."""
+
+ def _generateKeyValues(self, claims, title):
+ """Iterates through an item's claims, extracts the wikibaseProperties
and sends the item ID, property ID
+ and a relative score of 1 to stdout, separated by commas."""
+ ipPair = ipPairOld = None
+ for claim in claims:
+ statement = {i:j for i, j in self._pairwise(claim['m'])}
+ if statement == None: continue
+ try:
+ ipPair = "Q%s,P%s,1" % (str(title), str(statement['value'])) #
,1 is for relative score. It's same for all, ie. 1.
+ # This code may be added in the future to generate CSV rows
with values. Please make it in a different class:
+ ## if 'wikibase-entityid' not in statement: continue
+ ## value =
str(statement['wikibase-entityid']['numeric-id']).encode("utf-8",
'ignore').strip()
+ ## ipvPair = "Q%s,P%s----Q%s,1" % (str(title),
str(statement['value']), value) # for itemID,propertyID---valueID rows
+ ## pvPair = "P%s,Q%s,1" % (str(statement['value']), value) #
for propertyID,valueID rows
+ # After that, write ipvPair or pvPair to stdout, whichever is
needed.
+
+ if ipPairOld != ipPair: #This check is to prevent printing
duplicate itemID,propertyID pairs.
+ ipPairOld = ipPair
+ sys.stdout.write(ipPair.encode("utf-8", 'ignore').strip()
+ "\n")
+ except KeyError:
+ pass
+
+class RefClaimPropPairGeneratorMapper(AbstractWikiParserMapper):
+ """The mapper class for generating the claim guid-property pairs for
source refs using Hadoop."""
+
+ def _generateKeyValues(self, claims, title):
+ """Iterates through an item's claims, extracts the claim GUIDs and
their source ref wikibaseProperties
+ and sends them to stdout with claim GUID as key and prefixed property
ID as value."""
+ ipPair = ipPairOld = None
+ for claim in claims:
+ statement = {i:j for i, j in self._pairwise(claim['m'])}
+ if statement == None: continue
+ try:
+ guid = claim['g']
+ for ref in claim['refs']:
+ reference = {i:j for i, j in self._pairwise(ref[0])}
+ sys.stdout.write("%s,P%s,1\n" % (guid,
str(reference['value'])))
+ except KeyError:
+ pass
+
+class CountRefPropsMapper(AbstractWikiParserMapper):
+ """The mapper class to count frequencies of properties used in source refs
from the wikidata dump using Hadoop."""
+
+ def _generateKeyValues(self, claims, title):
+ """Iterates through an item's claims to extract the properties from
the source references and sends them
+ to stdout with the property id as the key and "1" as the value. This
is similar to MapReduce wordcount."""
+ for claim in claims:
+ if 'refs' not in claim: continue
+ for ref in claim['refs']:
+ source = {i:j for i, j in self._pairwise(ref[0])}
+ if source == None: continue
+ try:
+ prop = str(source['value'])
+ sys.stdout.write("P%s\t1\n" % prop)
+ except KeyError:
+ pass
+
+class CountGlobalPropsMapper(AbstractWikiParserMapper):
+ """The mapper class to count global property frequencies from the wikidata
dump using Hadoop."""
+
+ def _generateKeyValues(self, claims, title):
+ """Iterates through an item's claims to extract the properties from
the statements and sends them to stdout
+ with the property id as the key and "1" as the value. This is similar
to MapReduce wordcount."""
+ for claim in claims:
+ statement = {i:j for i, j in self._pairwise(claim['m'])}
+ if statement == None: continue
+ try:
+ prop = str(statement['value'])
+ sys.stdout.write("P%s\t1\n" % prop)
+ except KeyError:
+ pass
+
+class CountQualiferPropsMapper(AbstractWikiParserMapper):
+ """The mapper class to count global qualifier property frequencies from
the wikidata dump using Hadoop."""
+
+ def _generateKeyValues(self, claims, title):
+ """Iterates through an item's claims to extract the properties from
the statements and the properties of its
+ qualifiers if any exist, and sends them to stdout with the statement's
property id as the key and the qualifier
+ property id as the value. This is similar to MapReduce wordcount."""
+ for claim in claims:
+ statement = {i:j for i, j in self._pairwise(claim['m'])}
+ if statement == None: continue
+ try:
+ prop = str(statement['value'])
+ for q in claim['q']:
+ qualifier = {i:j for i, j in self._pairwise(q)}
+ sys.stdout.write("P%s\tP%s\n" % (prop,
str(qualifier['value'])))
+ except KeyError:
+ pass
+
+class CountPropertyValuesMapper(AbstractWikiParserMapper):
+ """The mapper class to count global qualifier property frequencies from
the wikidata dump using Hadoop."""
+
+ def _generateKeyValues(self, claims, title):
+ """Iterates through an item's claims to extract the wikibaseProperties
and wikibaseValues from the statements
+ and sends them to stdout with the wikibaseProperty id as the key and
the wikibaseValue as the value.
+ This is similar to MapReduce wordcount."""
+ for claim in claims:
+ statement = {i:j for i, j in self._pairwise(claim['m'])}
+ if statement == None: continue
+ try:
+ prop = str(statement['value'])
+ if 'wikibase-entityid' not in statement: continue
+ value =
str(statement['wikibase-entityid']['numeric-id']).encode("utf-8",
'ignore').strip()
+ sys.stdout.write("P%s\tQ%s\n" % (prop, value))
+ except KeyError:
+ pass
+
+if __name__ == '__main__':
+ main()
diff --git a/wikiparser/reducer.py b/wikiparser/reducer.py
new file mode 100644
index 0000000..4c3d1ab
--- /dev/null
+++ b/wikiparser/reducer.py
@@ -0,0 +1,142 @@
+#!/usr/bin/python
+
+"""This is the reducer module for the wikiparser Hadoop scripts. See main()
for usage info."""
+
+import sys
+
+def main():
+ """ Here are 6 types of sample commands:
+ 1. For generating the item-property pairs using Hadoop:
+ (Dataset used to suggest global claim properties)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/global-ip-pairs \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py global-ip-pairs' \
+ -reducer '/bin/cat'
+
+ 2. For generating the claim guid-property pairs for source refs using
Hadoop:
+ (Dataset used to suggest source ref properties)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/ref-cp-pairs \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py ref-cp-pairs' \
+ -reducer '/bin/cat'
+
+ 3. For counting frequencies of properties used in source refs:
+ (Dataset used to suggest source ref properties for empty source refs)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/ref-props \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py ref-props' \
+ -file /path/to/reducer.py -reducer '/path/to/reducer.py ref-props'
+
+ 4. For counting global property (used in all claims) frequencies:
+ (Dataset used to suggest global claim properties for empty items)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/global-props \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py global-props' \
+ -file /path/to/reducer.py -reducer '/path/to/reducer.py global-props'
+
+ 5. For counting frequency of properties used in qualifiers, to find the
top N popular qualifier properties for each property:
+ (Dataset used to suggest qualifiers, given a property)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/qual-props \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py qual-props' \
+ -file /path/to/reducer.py -reducer '/path/to/reducer.py qual-props'
+
+ 6. For counting frequency of values used with properties, to find the top
N popular values for each property:
+ (Dataset used to suggest values, given a property)
+ bin/hadoop jar contrib/streaming/hadoop*streaming*jar -libjars
/path/to/wikiparser-0.1.jar \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input /input/dump.xml -output /output/prop-values \
+ -file /path/to/mapper.py -mapper '/path/to/mapper.py prop-values' \
+ -file /path/to/reducer.py -reducer '/path/to/reducer.py prop-values'
+ """
+
+ reducerClass = {
+ 'ref-props': CountPropsReducer,
+ 'global-props': CountPropsReducer,
+ 'qual-props': CountQualiferPropsReducer,
+ 'prop-values': CountQualiferPropsReducer
+ }
+
+ try:
+ reducer = reducerClass[sys.argv[1]]()
+ except (KeyError, IndexError):
+ print "Unrecognized/missing reducer arguments."
+ print "Please enter any of ref-props, global-props, qual-props,
prop-values."
+ return
+
+ reducer.run()
+
+class BaseWikiParserReducer:
+ """Base/abstract class that should be inherited by all reducer classes"""
+
+ def run(self):
+ """Main driver method that iterates through the key-value pairs
obtained from the mapper and writes out the final results."""
+ raise NotImplementedError("Please implement this method in a
subclass.")
+
+class CountPropsReducer(BaseWikiParserReducer):
+ """The reducer class to count frequencies of properties used in source
refs from the wikidata dump, and global property frequencies using Hadoop."""
+
+ def run(self):
+ """Counts the number of occurrences for each property and outputs the
property id as key
+ and the occurrence frequency as the value to stdout.
+ """
+ oldKey = None
+ currentCount = 0
+ key = None
+
+ for i in sys.stdin:
+ (key, value) = i.split("\t")
+ try:
+ count = int(value.strip())
+ except ValueError:
+ continue
+
+ if oldKey == key:
+ currentCount += count
+
+ else:
+ if oldKey:
+ sys.stdout.write("%s,%d\n" % (oldKey, currentCount))
+ currentCount = count
+ oldKey = key
+
+ if oldKey == key:
+ sys.stdout.write("%s,%d\n" % (oldKey, currentCount))
+
+class CountQualiferPropsReducer(BaseWikiParserReducer):
+ """The reducer class to count global qualifier property frequencies from
the wikidata dump using Hadoop.
+ This is also used for counting the occurrences of values for each
property."""
+
+ def run(self):
+ """Counts the number of occurrences for each property and outputs the
property id as key
+ and the occurrence frequency as the value to stdout.
+ """
+ oldKey = None
+ key = None
+ qualifierCounts = {}
+
+ for i in sys.stdin:
+ (key, value) = i.split("\t")
+ value = value.strip()
+
+ if oldKey == key:
+ if value in qualifierCounts:
+ qualifierCounts[value] += 1
+ else:
+ qualifierCounts[value] = 1
+ else:
+ if oldKey:
+ sys.stdout.write("####%s\n" % oldKey) # The property ID,
followed by the list of qualifer prop counts
+ for q in qualifierCounts:
+ sys.stdout.write("%s,%d\n" % (q, qualifierCounts[q]))
+ oldKey = key
+ qualifierCounts = {}
+ qualifierCounts[value] = 1
+
+if __name__ == '__main__':
+ main()
diff --git a/wikiparser/runhadoop.sh b/wikiparser/runhadoop.sh
new file mode 100755
index 0000000..ab9d860
--- /dev/null
+++ b/wikiparser/runhadoop.sh
@@ -0,0 +1,35 @@
+#!/bin/bash
+ d=`dirname "$0"`
+
+ #You may need to change the following two variables to suit your
installation.
+ hadoop=/usr/share/hadoop
+ hadoop_executable=hadoop
+
+ if [ -z "$3" ]; then
+ echo "$0 <mode> <xmldump> <output> [<local-output>]"
+ echo "<mode> can be any of global-ip-pairs, ref-cp-pairs, ref-props,
global-props, qual-props, prop-values."
+ echo "<xmldump> is the location of the wikidata data dump file on
HDFS."
+ echo "<output> is the directory to write output to on HDFS."
+ echo "<local-output> is optional - if it is specified, the output from
HDFS will be copied to this file on the local file system."
+ exit 1
+ fi
+
+ if [ $1 == "global-ip-pairs" -o $1 == "ref-cp-pairs" ]; then
+ $hadoop_executable jar $hadoop/contrib/streaming/hadoop*streaming*.jar
\
+ -libjars "$d/target/wikiparser-0.1.jar" \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input "$2" -output "$3" \
+ -file "$d/mapper.py" -mapper "$d/mapper.py $1" \
+ -reducer '/bin/cat'
+ else
+ $hadoop_executable jar $hadoop/contrib/streaming/hadoop*streaming*.jar
\
+ -libjars "$d/target/wikiparser-0.1.jar" \
+ -inputformat
org.wikimedia.wikibase.entitysuggester.wikiparser.WikiPageInputFormat \
+ -input "$2" -output "$3" \
+ -file "$d/mapper.py" -mapper "$d/mapper.py $1" \
+ -file "$d/reducer.py" -reducer "$d/reducer.py $1"
+ fi
+
+ if [ -n "$4" ]; then
+ $hadoop_executable dfs -cat $3/part-* > $4
+ fi
diff --git a/wikiparser/wikiparser.py b/wikiparser/wikiparser.py
deleted file mode 100644
index c7e3731..0000000
--- a/wikiparser/wikiparser.py
+++ /dev/null
@@ -1,55 +0,0 @@
-#!/usr/bin/python
-
-from lxml import etree
-from StringIO import StringIO
-import json
-import sys
-
-page = ''
-
-def main():
- while True:
- page = ''
- i = sys.stdin.readline()
- if not i: break
- if '<page>' in i:
- i = sys.stdin.readline()
- while '</page>' not in i:
- page += i#.strip()
- i = sys.stdin.readline()
- page = '<page>' + page + '</page>'
- parsePage(page)
-
-def pairwise(iterable):
- from itertools import izip
- a = iter(iterable)
- return izip(a, a)
-
-def parsePage(page):
- tree = etree.parse(StringIO(page))
- page = {child.tag:child.text for child in tree.iter()}
- try:
- if page['ns'] == '0':
- title = page['title'][1:]
- text = json.loads(page['text'])
- statement = None
- if 'claims' in text:
- for a in text['claims']:
- statement = {i:j for i, j in pairwise(a['m'])}
- if statement != None:
- try:
- toyield1 = str(statement['value'])
- value =
str(statement['wikibase-entityid']['numeric-id']) if 'wikibase-entityid' in
statement else statement['string']
- toyield2 = str(statement['value']) + "----" + value
- sys.stdout.write("$$\t" + toyield1.encode("utf-8",
'ignore').strip() + "\n")
- sys.stdout.write("$$\t" + toyield2.encode("utf-8",
'ignore').strip() + "\n")
- sys.stdout.write("@@\t" + str(title) + "," +
toyield1.encode("utf-8", 'ignore').strip() + "\n")
- sys.stdout.write("@@\t" + str(title) + "," +
toyield2.encode("utf-8", 'ignore').strip() + "\n")
- except KeyError:
- pass
- except (KeyError, ValueError, TypeError) as e:
- sys.stderr.write("Error occurred for page : " + str(title) + ", ns = "
+ str(page['ns']) + "\n")
- sys.stderr.write(traceback.format_exc() + "\n")
-
-if __name__ == '__main__':
- main()
diff --git a/wikiparser/wikiparser_db.py b/wikiparser/wikiparser_db.py
deleted file mode 100644
index 958e9ab..0000000
--- a/wikiparser/wikiparser_db.py
+++ /dev/null
@@ -1,88 +0,0 @@
-#!/usr/bin/python
-
-from lxml import etree
-from StringIO import StringIO
-import json
-import sys
-import MySQLdb as mdb
-import traceback
-
-count = 0
-page = ''
-
-def main():
- con = None
- cur = None
- try:
- con = mdb.connect('localhost', 'root', 'password', 'wikidatawiki');
- cur = con.cursor()
- cur.execute("SET FOREIGN_KEY_CHECKS = 0")
- cur.execute("SET UNIQUE_CHECKS = 0")
- cur.execute("SET AUTOCOMMIT = 0")
- count = 0
- while True:
- page = ''
- i = sys.stdin.readline()
- if not i: break
- if '<page>' in i:
- i = sys.stdin.readline()
- while '</page>' not in i:
- page += i
- i = sys.stdin.readline()
- page = '<page>' + page + '</page>'
- parsePage(con, cur, page)
- count += 1
- if(count % 10000 == 0):
- con.commit()
- finally:
- if cur:
- cur.execute("SET UNIQUE_CHECKS = 1")
- cur.execute("SET FOREIGN_KEY_CHECKS = 1")
- con.commit()
- if con:
- con.close()
-
-def pairwise(iterable):
- from itertools import izip
- a = iter(iterable)
- return izip(a, a)
-
-def parsePage(con, cur, page):
- tree = etree.parse(StringIO(page))
- page = {child.tag:child.text for child in tree.iter()}
- try:
- if page['ns'] == '0':
- title = page['title'][1:]
- text = json.loads(page['text'])
- if 'en' not in text['label']: return
- label = text['label']['en'].encode("utf-8", 'ignore')
- cur.execute("""INSERT INTO label VALUES (%s, %s, %s)""",
(int(title), 'en', label))
- statement = None
- if 'claims' in text:
- for a in text['claims']:
- statement = {i:j for i, j in pairwise(a['m'])}
- if statement != None:
- try:
- toyield1 = str(statement['value'])
- value =
str(statement['wikibase-entityid']['numeric-id']) if 'wikibase-entityid' in
statement else statement['string']
- toyield2 = str(statement['value']) + "----" + value
- sys.stdout.write("$$\t" + toyield1.encode("utf-8",
'ignore').strip() + "\n")
- sys.stdout.write("$$\t" + toyield2.encode("utf-8",
'ignore').strip() + "\n")
- sys.stdout.write("@@\t" + str(title) + "," +
toyield1.encode("utf-8", 'ignore').strip() + "\n")
- sys.stdout.write("@@\t" + str(title) + "," +
toyield2.encode("utf-8", 'ignore').strip() + "\n")
- except KeyError:
- pass
- elif page['ns'] == '120':
- title = page['title'][10:]
- text = json.loads(page['text'])
- label = text['label']['en'].encode("utf-8", 'ignore')
- cur.execute("""INSERT INTO plabel VALUES (%s, %s, %s)""",
(int(title), 'en', label))
- except (KeyError, ValueError, TypeError) as e:
- sys.stderr.write("Error occurred for page : " + str(title) + ", ns = "
+ str(page['ns']) + "\n")
- sys.stderr.write(traceback.format_exc() + "\n")
- except mdb.Error, e:
- print "Error %d: %s" % (e.args[0],e.args[1])
- sys.exit(1)
-
-if __name__ == '__main__':
- main()
diff --git a/wikiparser/wikiparser_r.py b/wikiparser/wikiparser_r.py
deleted file mode 100644
index 0593c0e..0000000
--- a/wikiparser/wikiparser_r.py
+++ /dev/null
@@ -1,16 +0,0 @@
-#!/usr/bin/python
-
-import sys
-
-def main():
- listout = open(sys.argv[1], "w")
- for i in sys.stdin:
- (key, value) = i.split("\t")
- if key == "@@":
- sys.stdout.write(value)
- else:
- listout.write(value)
- listout.close()
-
-if __name__ == '__main__':
- main()
--
To view, visit https://gerrit.wikimedia.org/r/84463
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: merged
Gerrit-Change-Id: I33d8b0aa1927a07a08b405905f6e65a14632b116
Gerrit-PatchSet: 7
Gerrit-Project: mediawiki/extensions/WikidataEntitySuggester
Gerrit-Branch: master
Gerrit-Owner: Nilesh <[email protected]>
Gerrit-Reviewer: Addshore <[email protected]>
Gerrit-Reviewer: Daniel Kinzler <[email protected]>
Gerrit-Reviewer: Nilesh <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
