accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ujustgotbi...@apache.org
Subject [07/50] git commit: ACCUMULO-375 added code to separate map jobs for parsing wikipedia and for ingesting into Accumulo
Date Thu, 06 Feb 2014 05:39:46 GMT
ACCUMULO-375 added code to separate map jobs for parsing wikipedia and for ingesting into Accumulo

git-svn-id: https://svn.apache.org/repos/asf/incubator/accumulo/branches/1.4@1241579 13f79535-47bb-0310-9956-ffa450edef68


Project: http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/commit/a673727d
Tree: http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/tree/a673727d
Diff: http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/diff/a673727d

Branch: refs/heads/1.4.5-SNAPSHOT
Commit: a673727d6c09551d5a22b76bdbcb0dcc098573db
Parents: 686895f
Author: Adam Fuchs <afuchs@apache.org>
Authored: Tue Feb 7 19:48:39 2012 +0000
Committer: Adam Fuchs <afuchs@apache.org>
Committed: Tue Feb 7 19:48:39 2012 +0000

----------------------------------------------------------------------
 README.parallel                                 |  65 +++++
 ingest/bin/ingest_parallel.sh                   |  74 ++++++
 ingest/conf/wikipedia_parallel.xml.example      |  59 +++++
 .../wikisearch/ingest/ArticleExtractor.java     |  26 +-
 .../ingest/WikipediaConfiguration.java          |  17 ++
 .../ingest/WikipediaPartitionedIngester.java    | 247 +++++++++++++++++++
 .../ingest/WikipediaPartitionedMapper.java      | 192 ++++++++++++++
 .../wikisearch/ingest/WikipediaPartitioner.java | 108 ++++++++
 8 files changed, 787 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/blob/a673727d/README.parallel
----------------------------------------------------------------------
diff --git a/README.parallel b/README.parallel
new file mode 100644
index 0000000..18cab71
--- /dev/null
+++ b/README.parallel
@@ -0,0 +1,65 @@
+
+ This project contains a sample application for ingesting and querying wikipedia data.
+ 
+  
+ Ingest
+ ------
+ 
+ 	Prerequisites
+ 	-------------
+ 	1. Accumulo, Hadoop, and ZooKeeper must be installed and running
+ 	2. ACCUMULO_HOME and ZOOKEEPER_HOME must be defined in the environment
+ 	3. One or more wikipedia dump files (http://dumps.wikimedia.org/backup-index.html) placed
in an HDFS directory.
+	     You will want to grab the files with the link name of pages-articles.xml.bz2
+ 
+ 
+ 	INSTRUCTIONS
+ 	------------
+	1. Copy the ingest/conf/wikipedia_parallel.xml.example to ingest/conf/wikipedia.xml and
change it to specify Accumulo information. 
+	2. Copy the ingest/lib/wikisearch-*.jar and ingest/lib/protobuf*.jar to $ACCUMULO_HOME/lib/ext
+	3. Then run ingest/bin/ingest_parallel.sh with one argument (the name of the directory in
HDFS where the wikipedia XML 
+             files reside) and this will kick off a MapReduce job to ingest the data into
Accumulo.
+   
+ Query
+ -----
+ 
+ 	Prerequisites
+ 	-------------
+	1. The query software was tested using JBoss AS 6. Install this unless you feel like messing
with the installation.
+ 	
+	NOTE: Ran into a bug (https://issues.jboss.org/browse/RESTEASY-531) that did not allow an
EJB3.1 war file. The
+	workaround is to separate the RESTEasy servlet from the EJBs by creating an EJB jar and
a WAR file.
+	
+	INSTRUCTIONS
+	-------------
+	1. Copy the query/src/main/resources/META-INF/ejb-jar.xml.example file to 
+	   query/src/main/resources/META-INF/ejb-jar.xml. Modify to the file to contain the same

+	   information that you put into the wikipedia.xml file from the Ingest step above. 
+	2. Re-build the query distribution by running 'mvn package assembly:single' in the top-level
directory. 
+        3. Untar the resulting file in the $JBOSS_HOME/server/default directory.
+
+              $ cd $JBOSS_HOME/server/default
+              $ tar -xzf $ACCUMULO_HOME/src/examples/wikisearch/query/target/wikisearch-query*.tar.gz
+ 
+           This will place the dependent jars in the lib directory and the EJB jar into the
deploy directory.
+	4. Next, copy the wikisearch*.war file in the query-war/target directory to $JBOSS_HOME/server/default/deploy.

+	5. Start JBoss ($JBOSS_HOME/bin/run.sh)
+	6. Use the Accumulo shell and give the user permissions for the wikis that you loaded, for
example: 
+			setauths -u <user> -s all,enwiki,eswiki,frwiki,fawiki
+	7. Copy the following jars to the $ACCUMULO_HOME/lib/ext directory from the $JBOSS_HOME/server/default/lib
directory:
+	
+		commons-lang*.jar
+		kryo*.jar
+		minlog*.jar
+		commons-jexl*.jar
+		google-collections*.jar
+		
+	8. Copy the $JBOSS_HOME/server/default/deploy/wikisearch-query*.jar to $ACCUMULO_HOME/lib/ext.
+
+
+	9. At this point you should be able to open a browser and view the page: http://localhost:8080/accumulo-wikisearch/ui/ui.jsp.
+	You can issue the queries using this user interface or via the following REST urls: <host>/accumulo-wikisearch/rest/Query/xml,
+	<host>/accumulo-wikisearch/rest/Query/html, <host>/accumulo-wikisearch/rest/Query/yaml,
or <host>/accumulo-wikisearch/rest/Query/json.
+	There are two parameters to the REST service, query and auths. The query parameter is the
same string that you would type
+	into the search box at ui.jsp, and the auths parameter is a comma-separated list of wikis
that you want to search (i.e.
+	enwiki,frwiki,dewiki, etc. Or you can use all) 

http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/blob/a673727d/ingest/bin/ingest_parallel.sh
----------------------------------------------------------------------
diff --git a/ingest/bin/ingest_parallel.sh b/ingest/bin/ingest_parallel.sh
new file mode 100755
index 0000000..e921494
--- /dev/null
+++ b/ingest/bin/ingest_parallel.sh
@@ -0,0 +1,74 @@
+#!/bin/bash
+
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+
+THIS_SCRIPT="$0"
+SCRIPT_DIR="${THIS_SCRIPT%/*}"
+SCRIPT_DIR=`cd $SCRIPT_DIR ; pwd`
+echo $SCRIPT_DIR
+
+ACCUMULO_HOME=${ACCUMULO_HOME}
+ZOOKEEPER_HOME=${ZOOKEEPER_HOME}
+
+#
+# Check ZOOKEEPER_HOME
+#
+if [[ -z $ZOOKEEPER_HOME ]]; then
+	echo "You must set ZOOKEEPER_HOME environment variable"
+	exit -1;
+else
+	for f in $ZOOKEEPER_HOME/zookeeper-*.jar; do
+		CLASSPATH=$f
+		break
+	done	
+fi
+
+#
+# Check ACCUMULO_HOME
+#
+if [[ -z $ACCUMULO_HOME ]]; then
+	echo "You must set ACCUMULO_HOME environment variable"
+	exit -1;
+else
+	for f in $ACCUMULO_HOME/lib/*.jar; do
+		CLASSPATH=${CLASSPATH}:$f
+	done	
+fi
+
+#
+# Add our jars
+#
+for f in $SCRIPT_DIR/../lib/*.jar; do
+	CLASSPATH=${CLASSPATH}:$f  
+done
+
+#
+# Transform the classpath into a comma-separated list also
+#
+LIBJARS=`echo $CLASSPATH | sed 's/:/,/g'`
+
+
+#
+# Map/Reduce job
+#
+JAR=$SCRIPT_DIR/../lib/wikisearch-ingest-1.4.0-incubating-SNAPSHOT.jar
+CONF=$SCRIPT_DIR/../conf/wikipedia.xml
+HDFS_DATA_DIR=$1
+export HADOOP_CLASSPATH=$CLASSPATH
+echo "hadoop jar $JAR org.apache.accumulo.examples.wikisearch.ingest.WikipediaPartitionedIngester
-libjars $LIBJARS -conf $CONF -Dwikipedia.input=${HDFS_DATA_DIR}"
+hadoop jar $JAR org.apache.accumulo.examples.wikisearch.ingest.WikipediaPartitionedIngester
-libjars $LIBJARS -conf $CONF -Dwikipedia.input=${HDFS_DATA_DIR}

http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/blob/a673727d/ingest/conf/wikipedia_parallel.xml.example
----------------------------------------------------------------------
diff --git a/ingest/conf/wikipedia_parallel.xml.example b/ingest/conf/wikipedia_parallel.xml.example
new file mode 100644
index 0000000..cf20f01
--- /dev/null
+++ b/ingest/conf/wikipedia_parallel.xml.example
@@ -0,0 +1,59 @@
+<?xml version="1.0"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+-->
+<configuration>
+  <property>
+    <name>wikipedia.accumulo.zookeepers</name>
+    <value><!--zookeeper servers --></value>
+  </property>
+  <property>
+    <name>wikipedia.accumulo.instance_name</name>
+    <value><!--instance name --></value>
+  </property>
+  <property>
+    <name>wikipedia.accumulo.user</name>
+    <value><!--user name --></value>
+  </property>
+  <property>
+    <name>wikipedia.accumulo.password</name>
+    <value><!-- password --></value>
+  </property>
+  <property>
+    <name>wikipedia.accumulo.table</name>
+    <value><!--table name --></value>
+  </property>
+  <property>
+    <name>wikipedia.ingest.partitions</name>
+    <value><!--number of partitions --></value>
+  </property>
+  <property>
+    <name>wikipedia.partitioned.directory</name>
+    <value><!--hdfs directory for intemediate partitioned storage --></value>
+  </property>
+  <property>
+    <name>wikipedia.ingest.groups</name>
+    <value><!--the number of intermediate partition groups to generate --></value>
+  </property>
+  <property>
+    <name>wikipedia.run.partitioner</name>
+    <value><!--whether to run the partitioner step --></value>
+  </property>
+  <property>
+    <name>wikipedia.run.ingest</name>
+    <value><!--whether to run the ingest step --></value>
+  </property>
+</configuration>

http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/blob/a673727d/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/ArticleExtractor.java
----------------------------------------------------------------------
diff --git a/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/ArticleExtractor.java
b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/ArticleExtractor.java
index 54e47b6..06d1670 100644
--- a/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/ArticleExtractor.java
+++ b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/ArticleExtractor.java
@@ -16,6 +16,9 @@
  */
 package org.apache.accumulo.examples.wikisearch.ingest;
 
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
 import java.io.Reader;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
@@ -29,6 +32,7 @@ import javax.xml.stream.XMLStreamReader;
 
 import org.apache.accumulo.examples.wikisearch.normalizer.LcNoDiacriticsNormalizer;
 import org.apache.accumulo.examples.wikisearch.normalizer.NumberNormalizer;
+import org.apache.hadoop.io.Writable;
 
 
 public class ArticleExtractor {
@@ -37,13 +41,15 @@ public class ArticleExtractor {
   private static NumberNormalizer nn = new NumberNormalizer();
   private static LcNoDiacriticsNormalizer lcdn = new LcNoDiacriticsNormalizer();
   
-  public static class Article {
+  public static class Article implements Writable {
     int id;
     String title;
     long timestamp;
     String comments;
     String text;
     
+    public Article(){}
+    
     private Article(int id, String title, long timestamp, String comments, String text) {
       super();
       this.id = id;
@@ -90,6 +96,24 @@ public class ArticleExtractor {
       fields.put("COMMENTS", lcdn.normalizeFieldValue("COMMENTS", this.comments));
       return fields;
     }
+
+    @Override
+    public void readFields(DataInput in) throws IOException {
+      id = in.readInt();
+      title = in.readUTF();
+      timestamp = in.readLong();
+      comments = in.readUTF();
+      text = in.readUTF();
+    }
+
+    @Override
+    public void write(DataOutput out) throws IOException {
+      out.writeInt(id);
+      out.writeUTF(title);
+      out.writeLong(timestamp);
+      out.writeUTF(comments);
+      out.writeUTF(text);
+    }
     
   }
   

http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/blob/a673727d/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaConfiguration.java
----------------------------------------------------------------------
diff --git a/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaConfiguration.java
b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaConfiguration.java
index d76d713..5a0aad4 100644
--- a/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaConfiguration.java
+++ b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaConfiguration.java
@@ -48,6 +48,11 @@ public class WikipediaConfiguration {
 
   public final static String NUM_GROUPS = "wikipedia.ingest.groups";
 
+  public final static String PARTITIONED_ARTICLES_DIRECTORY = "wikipedia.partitioned.directory";
+  
+  public final static String RUN_PARTITIONER = "wikipedia.run.partitioner";
+  public final static String RUN_INGEST = "wikipedia.run.ingest";
+  
   
   public static String getUser(Configuration conf) {
     return conf.get(USER);
@@ -117,6 +122,18 @@ public class WikipediaConfiguration {
     return conf.getInt(NUM_GROUPS, 1);
   }
   
+  public static Path getPartitionedArticlesPath(Configuration conf) {
+    return new Path(conf.get(PARTITIONED_ARTICLES_DIRECTORY));
+  }
+  
+  public static boolean runPartitioner(Configuration conf) {
+    return conf.getBoolean(RUN_PARTITIONER, false);
+  }
+
+  public static boolean runIngest(Configuration conf) {
+    return conf.getBoolean(RUN_INGEST, true);
+  }
+
   /**
    * Helper method to get properties from Hadoop configuration
    * 

http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/blob/a673727d/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedIngester.java
----------------------------------------------------------------------
diff --git a/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedIngester.java
b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedIngester.java
new file mode 100644
index 0000000..e7493dc
--- /dev/null
+++ b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedIngester.java
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.examples.wikisearch.ingest;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.IteratorSetting.Column;
+import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.TableOperations;
+import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.iterators.user.SummingCombiner;
+import org.apache.accumulo.examples.wikisearch.ingest.ArticleExtractor.Article;
+import org.apache.accumulo.examples.wikisearch.iterator.GlobalIndexUidCombiner;
+import org.apache.accumulo.examples.wikisearch.iterator.TextIndexCombiner;
+import org.apache.accumulo.examples.wikisearch.reader.AggregatingRecordReader;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+
+public class WikipediaPartitionedIngester extends Configured implements Tool {
+  
+  public final static String INGEST_LANGUAGE = "wikipedia.ingest_language";
+  public final static String SPLIT_FILE = "wikipedia.split_file";
+  public final static String TABLE_NAME = "wikipedia.table";
+  
+  public static void main(String[] args) throws Exception {
+    int res = ToolRunner.run(new Configuration(), new WikipediaPartitionedIngester(), args);
+    System.exit(res);
+  }
+  
+  private void createTables(TableOperations tops, String tableName) throws AccumuloException,
AccumuloSecurityException, TableNotFoundException,
+      TableExistsException {
+    // Create the shard table
+    String indexTableName = tableName + "Index";
+    String reverseIndexTableName = tableName + "ReverseIndex";
+    String metadataTableName = tableName + "Metadata";
+    
+    // create the shard table
+    if (!tops.exists(tableName)) {
+      // Set a text index combiner on the given field names. No combiner is set if the option
is not supplied
+      String textIndexFamilies = WikipediaMapper.TOKENS_FIELD_NAME;
+      
+      tops.create(tableName);
+      if (textIndexFamilies.length() > 0) {
+        System.out.println("Adding content combiner on the fields: " + textIndexFamilies);
+        
+        IteratorSetting setting = new IteratorSetting(10, TextIndexCombiner.class);
+        List<Column> columns = new ArrayList<Column>();
+        for (String family : StringUtils.split(textIndexFamilies, ',')) {
+          columns.add(new Column("fi\0" + family));
+        }
+        TextIndexCombiner.setColumns(setting, columns);
+        TextIndexCombiner.setLossyness(setting, true);
+        
+        tops.attachIterator(tableName, setting, EnumSet.allOf(IteratorScope.class));
+      }
+      
+      // Set the locality group for the full content column family
+      tops.setLocalityGroups(tableName, Collections.singletonMap("WikipediaDocuments", Collections.singleton(new
Text(WikipediaMapper.DOCUMENT_COLUMN_FAMILY))));
+      
+    }
+    
+    if (!tops.exists(indexTableName)) {
+      tops.create(indexTableName);
+      // Add the UID combiner
+      IteratorSetting setting = new IteratorSetting(19, "UIDAggregator", GlobalIndexUidCombiner.class);
+      GlobalIndexUidCombiner.setCombineAllColumns(setting, true);
+      GlobalIndexUidCombiner.setLossyness(setting, true);
+      tops.attachIterator(indexTableName, setting, EnumSet.allOf(IteratorScope.class));
+    }
+    
+    if (!tops.exists(reverseIndexTableName)) {
+      tops.create(reverseIndexTableName);
+      // Add the UID combiner
+      IteratorSetting setting = new IteratorSetting(19, "UIDAggregator", GlobalIndexUidCombiner.class);
+      GlobalIndexUidCombiner.setCombineAllColumns(setting, true);
+      GlobalIndexUidCombiner.setLossyness(setting, true);
+      tops.attachIterator(reverseIndexTableName, setting, EnumSet.allOf(IteratorScope.class));
+    }
+    
+    if (!tops.exists(metadataTableName)) {
+      // Add the SummingCombiner with VARLEN encoding for the frequency column
+      tops.create(metadataTableName);
+      IteratorSetting setting = new IteratorSetting(10, SummingCombiner.class);
+      SummingCombiner.setColumns(setting, Collections.singletonList(new Column("f")));
+      SummingCombiner.setEncodingType(setting, SummingCombiner.Type.VARLEN);
+      tops.attachIterator(metadataTableName, setting, EnumSet.allOf(IteratorScope.class));
+    }
+  }
+  
+  @Override
+  public int run(String[] args) throws Exception {
+    Configuration conf = getConf();
+    if(WikipediaConfiguration.runPartitioner(conf))
+    {
+      int result = runPartitionerJob();
+      if(result != 0)
+        return result;
+    }
+    if(WikipediaConfiguration.runIngest(conf))
+      return runIngestJob();
+    return 0;
+  }
+  
+  public int runPartitionerJob() throws Exception
+  {
+    Job partitionerJob = new Job(getConf(), "Partition Wikipedia");
+    Configuration partitionerConf = partitionerJob.getConfiguration();
+    partitionerConf.set("mapred.map.tasks.speculative.execution", "false");
+
+    configurePartitionerJob(partitionerJob);
+    
+    List<Path> inputPaths = new ArrayList<Path>();
+    SortedSet<String> languages = new TreeSet<String>();
+    FileSystem fs = FileSystem.get(partitionerConf);
+    Path parent = new Path(partitionerConf.get("wikipedia.input"));
+    listFiles(parent, fs, inputPaths, languages);
+    
+    System.out.println("Input files in " + parent + ":" + inputPaths.size());
+    Path[] inputPathsArray = new Path[inputPaths.size()];
+    inputPaths.toArray(inputPathsArray);
+    
+    System.out.println("Languages:" + languages.size());
+
+    // setup input format
+    
+    WikipediaInputFormat.setInputPaths(partitionerJob, inputPathsArray);
+    
+    partitionerJob.setMapperClass(WikipediaPartitioner.class);
+    partitionerJob.setNumReduceTasks(0);
+
+    // setup output format
+    partitionerJob.setMapOutputKeyClass(Text.class);
+    partitionerJob.setMapOutputValueClass(Article.class);
+    partitionerJob.setOutputFormatClass(SequenceFileOutputFormat.class);
+    Path outputDir = WikipediaConfiguration.getPartitionedArticlesPath(partitionerConf);
+    SequenceFileOutputFormat.setOutputPath(partitionerJob, outputDir);
+    
+    return partitionerJob.waitForCompletion(true) ? 0 : 1;
+  }
+  
+  public int runIngestJob() throws Exception
+  {
+    Job ingestJob = new Job(getConf(), "Ingest Partitioned Wikipedia");
+    Configuration ingestConf = ingestJob.getConfiguration();
+    ingestConf.set("mapred.map.tasks.speculative.execution", "false");
+
+    String tablename = WikipediaConfiguration.getTableName(ingestConf);
+    
+    String zookeepers = WikipediaConfiguration.getZookeepers(ingestConf);
+    String instanceName = WikipediaConfiguration.getInstanceName(ingestConf);
+    
+    String user = WikipediaConfiguration.getUser(ingestConf);
+    byte[] password = WikipediaConfiguration.getPassword(ingestConf);
+    Connector connector = WikipediaConfiguration.getConnector(ingestConf);
+    
+    TableOperations tops = connector.tableOperations();
+    
+    createTables(tops, tablename);
+    
+    // setup input format
+    ingestJob.setInputFormatClass(SequenceFileInputFormat.class);
+    SequenceFileInputFormat.setInputPaths(ingestJob, WikipediaConfiguration.getPartitionedArticlesPath(ingestConf));
+
+    // setup output format
+    ingestJob.setMapOutputKeyClass(Text.class);
+    ingestJob.setMapOutputValueClass(Mutation.class);
+    ingestJob.setOutputFormatClass(AccumuloOutputFormat.class);
+    AccumuloOutputFormat.setOutputInfo(ingestJob.getConfiguration(), user, password, true,
tablename);
+    AccumuloOutputFormat.setZooKeeperInstance(ingestJob.getConfiguration(), instanceName,
zookeepers);
+    
+    return ingestJob.waitForCompletion(true) ? 0 : 1;
+  }
+  
+  public final static PathFilter partFilter = new PathFilter() {
+    @Override
+    public boolean accept(Path path) {
+      return path.getName().startsWith("part");
+    };
+  };
+  
+  protected void configurePartitionerJob(Job job) {
+    Configuration conf = job.getConfiguration();
+    job.setJarByClass(WikipediaPartitionedIngester.class);
+    job.setInputFormatClass(WikipediaInputFormat.class);
+    conf.set(AggregatingRecordReader.START_TOKEN, "<page>");
+    conf.set(AggregatingRecordReader.END_TOKEN, "</page>");
+  }
+  
+  protected static final Pattern filePattern = Pattern.compile("([a-z_]+).*.xml(.bz2)?");
+  
+  protected void listFiles(Path path, FileSystem fs, List<Path> files, Set<String>
languages) throws IOException {
+    for (FileStatus status : fs.listStatus(path)) {
+      if (status.isDir()) {
+        listFiles(status.getPath(), fs, files, languages);
+      } else {
+        Path p = status.getPath();
+        Matcher matcher = filePattern.matcher(p.getName());
+        if (matcher.matches()) {
+          languages.add(matcher.group(1));
+          files.add(p);
+        }
+      }
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/blob/a673727d/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedMapper.java
----------------------------------------------------------------------
diff --git a/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedMapper.java
b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedMapper.java
new file mode 100644
index 0000000..4d94c24
--- /dev/null
+++ b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedMapper.java
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/**
+ * 
+ */
+package org.apache.accumulo.examples.wikisearch.ingest;
+
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.nio.charset.Charset;
+import java.util.HashSet;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.accumulo.examples.wikisearch.ingest.ArticleExtractor.Article;
+import org.apache.accumulo.examples.wikisearch.normalizer.LcNoDiacriticsNormalizer;
+import org.apache.accumulo.examples.wikisearch.protobuf.Uid;
+import org.apache.accumulo.examples.wikisearch.protobuf.Uid.List.Builder;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.log4j.Logger;
+import org.apache.lucene.analysis.tokenattributes.TermAttribute;
+import org.apache.lucene.wikipedia.analysis.WikipediaTokenizer;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+
+public class WikipediaPartitionedMapper extends Mapper<Text,Article,Text,Mutation>
{
+  
+  private static final Logger log = Logger.getLogger(WikipediaPartitionedMapper.class);
+  
+  public final static Charset UTF8 = Charset.forName("UTF-8");
+  public static final String DOCUMENT_COLUMN_FAMILY = "d";
+  public static final String METADATA_EVENT_COLUMN_FAMILY = "e";
+  public static final String METADATA_INDEX_COLUMN_FAMILY = "i";
+  public static final String TOKENS_FIELD_NAME = "TEXT";
+  
+  private static final Value NULL_VALUE = new Value(new byte[0]);
+  private static final String cvPrefix = "all|";
+  
+  private int numPartitions = 0;
+
+  private Text tablename = null;
+  private Text indexTableName = null;
+  private Text reverseIndexTableName = null;
+  private Text metadataTableName = null;
+  
+  @Override
+  public void setup(Context context) {
+    Configuration conf = context.getConfiguration();
+    tablename = new Text(WikipediaConfiguration.getTableName(conf));
+    indexTableName = new Text(tablename + "Index");
+    reverseIndexTableName = new Text(tablename + "ReverseIndex");
+    metadataTableName = new Text(tablename + "Metadata");
+    
+    numPartitions = WikipediaConfiguration.getNumPartitions(conf);
+  }
+  
+  @Override
+  protected void map(Text language, Article article, Context context) throws IOException,
InterruptedException {
+    String NULL_BYTE = "\u0000";
+    String colfPrefix = language.toString() + NULL_BYTE;
+    String indexPrefix = "fi" + NULL_BYTE;
+    ColumnVisibility cv = new ColumnVisibility(cvPrefix + language);
+    
+    if (article != null) {
+      Text partitionId = new Text(Integer.toString(WikipediaMapper.getPartitionId(article,
numPartitions)));
+      
+      // Create the mutations for the document.
+      // Row is partition id, colf is language0articleid, colq is fieldName\0fieldValue
+      Mutation m = new Mutation(partitionId);
+      for (Entry<String,Object> entry : article.getFieldValues().entrySet()) {
+        m.put(colfPrefix + article.getId(), entry.getKey() + NULL_BYTE + entry.getValue().toString(),
cv, article.getTimestamp(), NULL_VALUE);
+        // Create mutations for the metadata table.
+        Mutation mm = new Mutation(entry.getKey());
+        mm.put(METADATA_EVENT_COLUMN_FAMILY, language.toString(), cv, article.getTimestamp(),
NULL_VALUE);
+        context.write(metadataTableName, mm);
+      }
+      
+      // Tokenize the content
+      Set<String> tokens = getTokens(article);
+      
+      // We are going to put the fields to be indexed into a multimap. This allows us to
iterate
+      // over the entire set once.
+      Multimap<String,String> indexFields = HashMultimap.create();
+      // Add the normalized field values
+      LcNoDiacriticsNormalizer normalizer = new LcNoDiacriticsNormalizer();
+      for (Entry<String,String> index : article.getNormalizedFieldValues().entrySet())
+        indexFields.put(index.getKey(), index.getValue());
+      // Add the tokens
+      for (String token : tokens)
+        indexFields.put(TOKENS_FIELD_NAME, normalizer.normalizeFieldValue("", token));
+      
+      for (Entry<String,String> index : indexFields.entries()) {
+        // Create mutations for the in partition index
+        // Row is partition id, colf is 'fi'\0fieldName, colq is fieldValue\0language\0article
id
+        m.put(indexPrefix + index.getKey(), index.getValue() + NULL_BYTE + colfPrefix + article.getId(),
cv, article.getTimestamp(), NULL_VALUE);
+        
+        // Create mutations for the global index
+        // Create a UID object for the Value
+        Builder uidBuilder = Uid.List.newBuilder();
+        uidBuilder.setIGNORE(false);
+        uidBuilder.setCOUNT(1);
+        uidBuilder.addUID(Integer.toString(article.getId()));
+        Uid.List uidList = uidBuilder.build();
+        Value val = new Value(uidList.toByteArray());
+        
+        // Create mutations for the global index
+        // Row is field value, colf is field name, colq is partitionid\0language, value is
Uid.List object
+        Mutation gm = new Mutation(index.getValue());
+        gm.put(index.getKey(), partitionId + NULL_BYTE + language, cv, article.getTimestamp(),
val);
+        context.write(indexTableName, gm);
+        
+        // Create mutations for the global reverse index
+        Mutation grm = new Mutation(StringUtils.reverse(index.getValue()));
+        grm.put(index.getKey(), partitionId + NULL_BYTE + language, cv, article.getTimestamp(),
val);
+        context.write(reverseIndexTableName, grm);
+        
+        // Create mutations for the metadata table.
+        Mutation mm = new Mutation(index.getKey());
+        mm.put(METADATA_INDEX_COLUMN_FAMILY, language + NULL_BYTE + LcNoDiacriticsNormalizer.class.getName(),
cv, article.getTimestamp(), NULL_VALUE);
+        context.write(metadataTableName, mm);
+        
+      }
+      // Add the entire text to the document section of the table.
+      // row is the partition, colf is 'd', colq is language\0articleid, value is Base64
encoded GZIP'd document
+      m.put(DOCUMENT_COLUMN_FAMILY, colfPrefix + article.getId(), cv, article.getTimestamp(),
new Value(Base64.encodeBase64(article.getText().getBytes())));
+      context.write(tablename, m);
+      
+    } else {
+      context.getCounter("wikipedia", "invalid articles").increment(1);
+    }
+    context.progress();
+  }
+  
+  /**
+   * Tokenize the wikipedia content
+   * 
+   * @param article
+   * @return
+   * @throws IOException
+   */
+  private Set<String> getTokens(Article article) throws IOException {
+    Set<String> tokenList = new HashSet<String>();
+    WikipediaTokenizer tok = new WikipediaTokenizer(new StringReader(article.getText()));
+    TermAttribute term = tok.addAttribute(TermAttribute.class);
+    try {
+      while (tok.incrementToken()) {
+        String token = term.term();
+        if (!StringUtils.isEmpty(token))
+          tokenList.add(token);
+      }
+    } catch (IOException e) {
+      log.error("Error tokenizing text", e);
+    } finally {
+      try {
+        tok.end();
+      } catch (IOException e) {
+        log.error("Error calling end()", e);
+      } finally {
+        try {
+          tok.close();
+        } catch (IOException e) {
+          log.error("Error closing tokenizer", e);
+        }
+      }
+    }
+    return tokenList;
+  }
+  
+}

http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/blob/a673727d/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitioner.java
----------------------------------------------------------------------
diff --git a/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitioner.java
b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitioner.java
new file mode 100644
index 0000000..82af9fd
--- /dev/null
+++ b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitioner.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/**
+ * 
+ */
+package org.apache.accumulo.examples.wikisearch.ingest;
+
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.StringReader;
+import java.nio.charset.Charset;
+import java.util.HashSet;
+import java.util.IllegalFormatException;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.accumulo.examples.wikisearch.ingest.ArticleExtractor.Article;
+import org.apache.accumulo.examples.wikisearch.ingest.WikipediaInputFormat.WikipediaInputSplit;
+import org.apache.accumulo.examples.wikisearch.normalizer.LcNoDiacriticsNormalizer;
+import org.apache.accumulo.examples.wikisearch.protobuf.Uid;
+import org.apache.accumulo.examples.wikisearch.protobuf.Uid.List.Builder;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+import org.apache.log4j.Logger;
+import org.apache.lucene.analysis.tokenattributes.TermAttribute;
+import org.apache.lucene.wikipedia.analysis.WikipediaTokenizer;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+
+public class WikipediaPartitioner extends Mapper<LongWritable,Text,Text,Article> {
+  
+  private static final Logger log = Logger.getLogger(WikipediaPartitioner.class);
+  
+  public final static Charset UTF8 = Charset.forName("UTF-8");
+  public static final String DOCUMENT_COLUMN_FAMILY = "d";
+  public static final String METADATA_EVENT_COLUMN_FAMILY = "e";
+  public static final String METADATA_INDEX_COLUMN_FAMILY = "i";
+  public static final String TOKENS_FIELD_NAME = "TEXT";
+  
+  private final static Pattern languagePattern = Pattern.compile("([a-z_]+).*.xml(.bz2)?");
+  
+  private ArticleExtractor extractor;
+  private String language;
+
+  private int myGroup = -1;
+  private int numGroups = -1;
+  
+  @Override
+  public void setup(Context context) {
+    Configuration conf = context.getConfiguration();
+    
+    WikipediaInputSplit wiSplit = (WikipediaInputSplit)context.getInputSplit();
+    myGroup = wiSplit.getPartition();
+    numGroups = WikipediaConfiguration.getNumGroups(conf);
+    
+    FileSplit split = wiSplit.getFileSplit();
+    String fileName = split.getPath().getName();
+    Matcher matcher = languagePattern.matcher(fileName);
+    if (matcher.matches()) {
+      language = matcher.group(1).replace('_', '-').toLowerCase();
+    } else {
+      throw new RuntimeException("Unknown ingest language! " + fileName);
+    }
+    extractor = new ArticleExtractor();
+  }
+  
+  @Override
+  protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
{
+    Article article = extractor.extract(new InputStreamReader(new ByteArrayInputStream(value.getBytes()),
UTF8));
+    if (article != null) {
+      int groupId = WikipediaMapper.getPartitionId(article, numGroups);
+      if(groupId != myGroup)
+        return;
+      context.write(new Text(language), article);
+    } else {
+      context.getCounter("wikipedia", "invalid articles").increment(1);
+      context.progress();
+    }
+  }
+  
+}


Mime
View raw message