| /**
|
| * 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.hadoop.contrib.index.main;
|
|
|
| import java.io.IOException;
|
| import java.text.NumberFormat;
|
| import java.util.Arrays;
|
|
|
| import org.apache.commons.logging.Log;
|
| import org.apache.commons.logging.LogFactory;
|
| import org.apache.hadoop.conf.Configuration;
|
| import org.apache.hadoop.contrib.index.mapred.IndexUpdateConfiguration;
|
| import org.apache.hadoop.contrib.index.mapred.IIndexUpdater;
|
| import org.apache.hadoop.contrib.index.mapred.Shard;
|
| import org.apache.hadoop.fs.FileStatus;
|
| import org.apache.hadoop.fs.FileSystem;
|
| import org.apache.hadoop.fs.Path;
|
| import org.apache.hadoop.mapred.FileOutputFormat; |
| import org.apache.hadoop.mapred.JobConf;
|
| import org.apache.hadoop.mapred.FileInputFormat;
|
| import org.apache.hadoop.util.ReflectionUtils;
|
|
|
| /**
|
| * A distributed "index" is partitioned into "shards". Each shard corresponds
|
| * to a Lucene instance. This class contains the main() method which uses a
|
| * Map/Reduce job to analyze documents and update Lucene instances in parallel.
|
| *
|
| * The main() method in UpdateIndex requires the following information for
|
| * updating the shards:
|
| * - Input formatter. This specifies how to format the input documents.
|
| * - Analysis. This defines the analyzer to use on the input. The analyzer
|
| * determines whether a document is being inserted, updated, or deleted.
|
| * For inserts or updates, the analyzer also converts each input document
|
| * into a Lucene document.
|
| * - Input paths. This provides the location(s) of updated documents,
|
| * e.g., HDFS files or directories, or HBase tables.
|
| * - Shard paths, or index path with the number of shards. Either specify
|
| * the path for each shard, or specify an index path and the shards are
|
| * the sub-directories of the index directory.
|
| * - Output path. When the update to a shard is done, a message is put here.
|
| * - Number of map tasks.
|
| *
|
| * All of the information can be specified in a configuration file. All but
|
| * the first two can also be specified as command line options. Check out
|
| * conf/index-config.xml.template for other configurable parameters.
|
| *
|
| * Note: Because of the parallel nature of Map/Reduce, the behaviour of
|
| * multiple inserts, deletes or updates to the same document is undefined.
|
| */
|
| public class UpdateIndex {
|
| public static final Log LOG = LogFactory.getLog(UpdateIndex.class);
|
|
|
| private static final NumberFormat NUMBER_FORMAT = NumberFormat.getInstance();
|
| static {
|
| NUMBER_FORMAT.setMinimumIntegerDigits(5);
|
| NUMBER_FORMAT.setGroupingUsed(false);
|
| }
|
|
|
| private static long now() {
|
| return System.currentTimeMillis();
|
| }
|
|
|
| private static void printUsage(String cmd) {
|
| System.err.println("Usage: java " + UpdateIndex.class.getName() + "\n"
|
| + " -inputPaths <inputPath,inputPath>\n"
|
| + " -outputPath <outputPath>\n"
|
| + " -shards <shardDir,shardDir>\n"
|
| + " -indexPath <indexPath>\n"
|
| + " -numShards <num>\n"
|
| + " -numMapTasks <num>\n"
|
| + " -conf <confPath>\n"
|
| + "Note: Do not use both -shards option and -indexPath option.");
|
| }
|
|
|
| private static String getIndexPath(Configuration conf) {
|
| return conf.get("sea.index.path");
|
| }
|
|
|
| private static int getNumShards(Configuration conf) {
|
| return conf.getInt("sea.num.shards", 1);
|
| }
|
|
|
| private static Shard[] createShards(String indexPath, int numShards,
|
| Configuration conf) throws IOException {
|
|
|
| String parent = Shard.normalizePath(indexPath) + Path.SEPARATOR;
|
| long versionNumber = -1;
|
| long generation = -1;
|
|
|
| FileSystem fs = FileSystem.get(conf);
|
| Path path = new Path(indexPath);
|
|
|
| if (fs.exists(path)) {
|
| FileStatus[] fileStatus = fs.listStatus(path);
|
| String[] shardNames = new String[fileStatus.length];
|
| int count = 0;
|
| for (int i = 0; i < fileStatus.length; i++) {
|
| if (fileStatus[i].isDirectory()) {
|
| shardNames[count] = fileStatus[i].getPath().getName();
|
| count++;
|
| }
|
| }
|
| Arrays.sort(shardNames, 0, count);
|
|
|
| Shard[] shards = new Shard[count >= numShards ? count : numShards];
|
| for (int i = 0; i < count; i++) {
|
| shards[i] =
|
| new Shard(versionNumber, parent + shardNames[i], generation);
|
| }
|
|
|
| int number = count;
|
| for (int i = count; i < numShards; i++) {
|
| String shardPath;
|
| while (true) {
|
| shardPath = parent + NUMBER_FORMAT.format(number++);
|
| if (!fs.exists(new Path(shardPath))) {
|
| break;
|
| }
|
| }
|
| shards[i] = new Shard(versionNumber, shardPath, generation);
|
| }
|
| return shards;
|
| } else {
|
| Shard[] shards = new Shard[numShards];
|
| for (int i = 0; i < shards.length; i++) {
|
| shards[i] =
|
| new Shard(versionNumber, parent + NUMBER_FORMAT.format(i),
|
| generation);
|
| }
|
| return shards;
|
| }
|
| }
|
|
|
| /**
|
| * The main() method
|
| * @param argv
|
| */
|
| public static void main(String[] argv) {
|
| if (argv.length == 0) {
|
| printUsage("");
|
| System.exit(-1);
|
| }
|
|
|
| String inputPathsString = null;
|
| Path outputPath = null;
|
| String shardsString = null;
|
| String indexPath = null;
|
| int numShards = -1;
|
| int numMapTasks = -1;
|
| Configuration conf = new Configuration();
|
| String confPath = null;
|
|
|
| // parse the command line
|
| for (int i = 0; i < argv.length; i++) { // parse command line
|
| if (argv[i].equals("-inputPaths")) {
|
| inputPathsString = argv[++i];
|
| } else if (argv[i].equals("-outputPath")) {
|
| outputPath = new Path(argv[++i]);
|
| } else if (argv[i].equals("-shards")) {
|
| shardsString = argv[++i];
|
| } else if (argv[i].equals("-indexPath")) {
|
| indexPath = argv[++i];
|
| } else if (argv[i].equals("-numShards")) {
|
| numShards = Integer.parseInt(argv[++i]);
|
| } else if (argv[i].equals("-numMapTasks")) {
|
| numMapTasks = Integer.parseInt(argv[++i]);
|
| } else if (argv[i].equals("-conf")) {
|
| // add as a local FS resource
|
| confPath = argv[++i];
|
| conf.addResource(new Path(confPath));
|
| } else {
|
| System.out.println("Unknown option " + argv[i] + " w/ value "
|
| + argv[++i]);
|
| }
|
| }
|
| LOG.info("inputPaths = " + inputPathsString);
|
| LOG.info("outputPath = " + outputPath);
|
| LOG.info("shards = " + shardsString);
|
| LOG.info("indexPath = " + indexPath);
|
| LOG.info("numShards = " + numShards);
|
| LOG.info("numMapTasks= " + numMapTasks);
|
| LOG.info("confPath = " + confPath);
|
|
|
| Path[] inputPaths = null;
|
| Shard[] shards = null;
|
|
|
| JobConf jobConf = new JobConf(conf);
|
| IndexUpdateConfiguration iconf = new IndexUpdateConfiguration(jobConf);
|
|
|
| if (inputPathsString != null) {
|
| jobConf.set(org.apache.hadoop.mapreduce.lib.input.
|
| FileInputFormat.INPUT_DIR, inputPathsString);
|
| }
|
| inputPaths = FileInputFormat.getInputPaths(jobConf);
|
| if (inputPaths.length == 0) {
|
| inputPaths = null;
|
| }
|
|
|
| if (outputPath == null) {
|
| outputPath = FileOutputFormat.getOutputPath(jobConf); |
| }
|
|
|
| if (inputPaths == null || outputPath == null) {
|
| System.err.println("InputPaths and outputPath must be specified.");
|
| printUsage("");
|
| System.exit(-1);
|
| }
|
|
|
| if (shardsString != null) {
|
| iconf.setIndexShards(shardsString);
|
| }
|
| shards = Shard.getIndexShards(iconf);
|
| if (shards != null && shards.length == 0) {
|
| shards = null;
|
| }
|
|
|
| if (indexPath == null) {
|
| indexPath = getIndexPath(conf);
|
| }
|
| if (numShards <= 0) {
|
| numShards = getNumShards(conf);
|
| }
|
|
|
| if (shards == null && indexPath == null) {
|
| System.err.println("Either shards or indexPath must be specified.");
|
| printUsage("");
|
| System.exit(-1);
|
| }
|
|
|
| if (numMapTasks <= 0) {
|
| numMapTasks = jobConf.getNumMapTasks();
|
| }
|
|
|
| try {
|
| // create shards and set their directories if necessary
|
| if (shards == null) {
|
| shards = createShards(indexPath, numShards, conf);
|
| }
|
|
|
| long startTime = now();
|
| try {
|
| IIndexUpdater updater =
|
| (IIndexUpdater) ReflectionUtils.newInstance(
|
| iconf.getIndexUpdaterClass(), conf);
|
| LOG.info("sea.index.updater = "
|
| + iconf.getIndexUpdaterClass().getName());
|
|
|
| updater.run(conf, inputPaths, outputPath, numMapTasks, shards);
|
| LOG.info("Index update job is done");
|
|
|
| } finally {
|
| long elapsedTime = now() - startTime;
|
| LOG.info("Elapsed time is " + (elapsedTime / 1000) + "s");
|
| System.out.println("Elapsed time is " + (elapsedTime / 1000) + "s");
|
| }
|
| } catch (Exception e) {
|
| e.printStackTrace(System.err);
|
| }
|
| }
|
| }
|
