Updated Branches: refs/heads/master ce0cfaea2 -> 4fc43e954
The CSV loader now works as a CLI program. You can invoke it via $BLUR_HOME/bin/blur csv arg1 arg2 ... Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/f564924c Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/f564924c Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/f564924c Branch: refs/heads/master Commit: f564924c3d6458bcec32faf5beadfbc2260d9ba7 Parents: ce0cfae Author: Aaron McCurry <[email protected]> Authored: Fri Aug 23 04:41:12 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Fri Aug 23 04:41:12 2013 -0400 ---------------------------------------------------------------------- .../blur/mapreduce/lib/BlurMapReduceUtil.java | 226 +++++++++++++++++++ .../blur/mapreduce/lib/BlurOutputFormat.java | 16 +- .../blur/mapreduce/lib/CsvBlurDriver.java | 29 ++- .../blur/mapreduce/lib/CsvBlurMapper.java | 37 +-- distribution/src/main/scripts/bin/blur | 6 + .../src/main/scripts/bin/blur-config.sh | 4 + 6 files changed, 282 insertions(+), 36 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/f564924c/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurMapReduceUtil.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurMapReduceUtil.java b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurMapReduceUtil.java new file mode 100644 index 0000000..5ee26eb --- /dev/null +++ b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurMapReduceUtil.java @@ -0,0 +1,226 @@ +/** + * 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.blur.mapreduce.lib; + +import java.io.IOException; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.net.URL; +import java.net.URLDecoder; +import java.util.ArrayList; +import java.util.Enumeration; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import org.apache.blur.log.Log; +import org.apache.blur.log.LogFactory; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.util.StringUtils; + +/** + * This utility code was taken from HBase to locate classes and the jars files + * to add to the MapReduce Job. + */ +public class BlurMapReduceUtil { + + private final static Log LOG = LogFactory.getLog(BlurMapReduceUtil.class); + + /** + * Add the Blur dependency jars as well as jars for any of the configured job + * classes to the job configuration, so that JobClient will ship them to the + * cluster and add them to the DistributedCache. + */ + public static void addDependencyJars(Job job) throws IOException { + try { + addDependencyJars(job.getConfiguration(), org.apache.zookeeper.ZooKeeper.class, job.getMapOutputKeyClass(), + job.getMapOutputValueClass(), job.getInputFormatClass(), job.getOutputKeyClass(), job.getOutputValueClass(), + job.getOutputFormatClass(), job.getPartitionerClass(), job.getCombinerClass()); + addAllJarsInBlurLib(job.getConfiguration()); + } catch (ClassNotFoundException e) { + throw new IOException(e); + } + } + + /** + * Adds all the jars in the same path as the blur jar files. + * @param conf + * @throws IOException + */ + public static void addAllJarsInBlurLib(Configuration conf) throws IOException { + FileSystem localFs = FileSystem.getLocal(conf); + Set<String> jars = new HashSet<String>(); + jars.addAll(conf.getStringCollection("tmpjars")); + + String property = System.getProperty("java.class.path"); + String[] files = property.split("\\:"); + + String blurLibPath = getPath("blur-", files); + if (blurLibPath == null) { + return; + } + List<String> pathes = getPathes(blurLibPath, files); + for (String pathStr : pathes) { + Path path = new Path(pathStr); + if (!localFs.exists(path)) { + LOG.warn("Could not validate jar file " + path); + continue; + } + jars.add(path.makeQualified(localFs).toString()); + } + if (jars.isEmpty()) { + return; + } + conf.set("tmpjars", StringUtils.arrayToString(jars.toArray(new String[0]))); + } + + private static List<String> getPathes(String path, String[] files) { + List<String> pathes = new ArrayList<String>(); + for (String file : files) { + if (file.startsWith(path)) { + pathes.add(file); + } + } + return pathes; + } + + private static String getPath(String startsWith, String[] files) { + for (String file : files) { + int lastIndexOf = file.lastIndexOf('/'); + String fileName = file.substring(lastIndexOf + 1); + if (fileName.startsWith(startsWith)) { + return file.substring(0, lastIndexOf); + } + } + return null; + } + + /** + * Add the jars containing the given classes to the job's configuration such + * that JobClient will ship them to the cluster and add them to the + * DistributedCache. + */ + public static void addDependencyJars(Configuration conf, Class<?>... classes) throws IOException { + FileSystem localFs = FileSystem.getLocal(conf); + Set<String> jars = new HashSet<String>(); + // Add jars that are already in the tmpjars variable + jars.addAll(conf.getStringCollection("tmpjars")); + + // Add jars containing the specified classes + for (Class<?> clazz : classes) { + if (clazz == null) { + continue; + } + + String pathStr = findOrCreateJar(clazz); + if (pathStr == null) { + LOG.warn("Could not find jar for class " + clazz + " in order to ship it to the cluster."); + continue; + } + Path path = new Path(pathStr); + if (!localFs.exists(path)) { + LOG.warn("Could not validate jar file " + path + " for class " + clazz); + continue; + } + jars.add(path.makeQualified(localFs).toString()); + } + if (jars.isEmpty()) { + return; + } + + conf.set("tmpjars", StringUtils.arrayToString(jars.toArray(new String[0]))); + } + + /** + * If org.apache.hadoop.util.JarFinder is available (0.23+ hadoop), finds the + * Jar for a class or creates it if it doesn't exist. If the class is in a + * directory in the classpath, it creates a Jar on the fly with the contents + * of the directory and returns the path to that Jar. If a Jar is created, it + * is created in the system temporary directory. + * + * Otherwise, returns an existing jar that contains a class of the same name. + * + * @param my_class + * the class to find. + * @return a jar file that contains the class, or null. + * @throws IOException + */ + private static String findOrCreateJar(Class<?> my_class) throws IOException { + try { + Class<?> jarFinder = Class.forName("org.apache.hadoop.util.JarFinder"); + // hadoop-0.23 has a JarFinder class that will create the jar + // if it doesn't exist. Note that this is needed to run the mapreduce + // unit tests post-0.23, because mapreduce v2 requires the relevant jars + // to be in the mr cluster to do output, split, etc. At unit test time, + // the hbase jars do not exist, so we need to create some. Note that we + // can safely fall back to findContainingJars for pre-0.23 mapreduce. + Method m = jarFinder.getMethod("getJar", Class.class); + return (String) m.invoke(null, my_class); + } catch (InvocationTargetException ite) { + // function was properly called, but threw it's own exception + throw new IOException(ite.getCause()); + } catch (Exception e) { + // ignore all other exceptions. related to reflection failure + } + + LOG.debug("New JarFinder: org.apache.hadoop.util.JarFinder.getJar " + "not available. Using old findContainingJar"); + return findContainingJar(my_class); + } + + /** + * Find a jar that contains a class of the same name, if any. It will return a + * jar file, even if that is not the first thing on the class path that has a + * class with the same name. + * + * This is shamelessly copied from JobConf + * + * @param my_class + * the class to find. + * @return a jar file that contains the class, or null. + * @throws IOException + */ + private static String findContainingJar(Class<?> my_class) { + ClassLoader loader = my_class.getClassLoader(); + String class_file = my_class.getName().replaceAll("\\.", "/") + ".class"; + try { + for (Enumeration<URL> itr = loader.getResources(class_file); itr.hasMoreElements();) { + URL url = itr.nextElement(); + if ("jar".equals(url.getProtocol())) { + String toReturn = url.getPath(); + if (toReturn.startsWith("file:")) { + toReturn = toReturn.substring("file:".length()); + } + // URLDecoder is a misnamed class, since it actually decodes + // x-www-form-urlencoded MIME type rather than actual + // URL encoding (which the file path has). Therefore it would + // decode +s to ' 's which is incorrect (spaces are actually + // either unencoded or encoded as "%20"). Replace +s first, so + // that they are kept sacred during the decoding process. + toReturn = toReturn.replaceAll("\\+", "%2B"); + toReturn = URLDecoder.decode(toReturn, "UTF-8"); + return toReturn.replaceAll("!.*$", ""); + } + } + } catch (IOException e) { + throw new RuntimeException(e); + } + return null; + } +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/f564924c/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java index 78b0d0f..2f59a20 100644 --- a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java +++ b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java @@ -376,8 +376,7 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { private ProgressableDirectory _localTmpDir; private String _deletedRowId; - public BlurRecordWriter(Configuration configuration, int attemptId, String tmpDirName) - throws IOException { + public BlurRecordWriter(Configuration configuration, int attemptId, String tmpDirName) throws IOException { _indexLocally = BlurOutputFormat.isIndexLocally(configuration); _optimizeInFlight = BlurOutputFormat.isOptimizeInFlight(configuration); @@ -394,11 +393,11 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { _finalDir = new ProgressableDirectory(new HdfsDirectory(configuration, _newIndex), BlurOutputFormat.getProgressable()); _finalDir.setLockFactory(NoLockFactory.getNoLockFactory()); - + TableContext tableContext = TableContext.create(tableDescriptor); _fieldManager = tableContext.getFieldManager(); Analyzer analyzer = _fieldManager.getAnalyzerForIndex(); - + _conf = new IndexWriterConfig(LuceneVersionConstant.LUCENE_VERSION, analyzer); TieredMergePolicy mergePolicy = (TieredMergePolicy) _conf.getMergePolicy(); mergePolicy.setUseCompoundFile(false); @@ -459,7 +458,7 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { return; } _columnCount.increment(record.getColumns().size()); - List<Field> document = TransactionRecorder.getDoc(_fieldManager,blurRecord.getRowId(), record); + List<Field> document = TransactionRecorder.getDoc(_fieldManager, blurRecord.getRowId(), record); List<Field> dup = _documents.put(recordId, document); if (dup != null) { _recordDuplicateCount.increment(1); @@ -486,8 +485,9 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { _localTmpPath = new File(localDirPath, UUID.randomUUID().toString() + ".tmp"); _localTmpDir = new ProgressableDirectory(FSDirectory.open(_localTmpPath), BlurOutputFormat.getProgressable()); _localTmpWriter = new IndexWriter(_localTmpDir, _overFlowConf.clone()); - //The local tmp writer has merging disabled so the first document in is going to be doc 0. - //Therefore the first document added is the prime doc + // The local tmp writer has merging disabled so the first document in is + // going to be doc 0. + // Therefore the first document added is the prime doc List<List<Field>> docs = new ArrayList<List<Field>>(_documents.values()); docs.get(0).add(new StringField(BlurConstants.PRIME_DOC, BlurConstants.PRIME_DOC_VALUE, Store.NO)); _localTmpWriter.addDocuments(docs); @@ -619,6 +619,8 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { job.setOutputValueClass(BlurMutate.class); job.setOutputFormatClass(BlurOutputFormat.class); setTableDescriptor(job, tableDescriptor); + BlurMapReduceUtil.addDependencyJars(job); + BlurMapReduceUtil.addAllJarsInBlurLib(job.getConfiguration()); } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/f564924c/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/CsvBlurDriver.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/CsvBlurDriver.java b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/CsvBlurDriver.java index e388a65..48bb902 100644 --- a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/CsvBlurDriver.java +++ b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/CsvBlurDriver.java @@ -55,6 +55,8 @@ import org.apache.hadoop.mapreduce.lib.input.SequenceFileRecordReader; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.util.GenericOptionsParser; +import com.google.common.base.Splitter; + @SuppressWarnings("static-access") public class CsvBlurDriver { @@ -176,7 +178,9 @@ public class CsvBlurDriver { return null; } for (String p : getSubArray(values, 1)) { - CsvBlurMapper.addFamilyPath(job, values[0], new Path(p)); + Path path = new Path(p); + CsvBlurMapper.addFamilyPath(job, values[0], path); + FileInputFormat.addInputPath(job, path); } } } @@ -212,6 +216,7 @@ public class CsvBlurDriver { } } BlurOutputFormat.setupJob(job, tableDescriptor); + BlurMapReduceUtil.addDependencyJars(job.getConfiguration(), Splitter.class); return job; } @@ -256,16 +261,18 @@ public class CsvBlurDriver { .withDescription("The directory to index. (hdfs://namenode/input/in1)").create("i")); options.addOption(OptionBuilder.withArgName("family path*").hasArgs() .withDescription("The directory to index with family name. (family hdfs://namenode/input/in1)").create("I")); - options.addOption(OptionBuilder - .withArgName("auto generate record ids") - .withDescription( - "No Record Ids - Automatically generate record ids for each record based on a MD5 has of the data within the record.") - .create("a")); - options.addOption(OptionBuilder - .withArgName("auto generate row ids") - .withDescription( - "No Row Ids - Automatically generate row ids for each record based on a MD5 has of the data within the record.") - .create("A")); + options + .addOption(OptionBuilder + .withArgName("auto generate record ids") + .withDescription( + "No Record Ids - Automatically generate record ids for each record based on a MD5 has of the data within the record.") + .create("a")); + options + .addOption(OptionBuilder + .withArgName("auto generate row ids") + .withDescription( + "No Row Ids - Automatically generate row ids for each record based on a MD5 has of the data within the record.") + .create("A")); options.addOption(OptionBuilder.withArgName("disable optimize indexes during copy") .withDescription("Disable optimize indexes during copy, this has very little overhead. (enabled by default)") .create("o")); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/f564924c/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/CsvBlurMapper.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/CsvBlurMapper.java b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/CsvBlurMapper.java index 645af5d..eeadb20 100644 --- a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/CsvBlurMapper.java +++ b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/CsvBlurMapper.java @@ -164,7 +164,7 @@ public class CsvBlurMapper extends BaseBlurMapper<Writable, Text> { * boolean. */ public static void setAutoGenerateRowIdAsHashOfData(Job job, boolean autoGenerateRowIdAsHashOfData) { - setAutoGenerateRecordIdAsHashOfData(job.getConfiguration(), autoGenerateRowIdAsHashOfData); + setAutoGenerateRowIdAsHashOfData(job.getConfiguration(), autoGenerateRowIdAsHashOfData); } /** @@ -422,23 +422,21 @@ public class CsvBlurMapper extends BaseBlurMapper<Writable, Text> { throw new IOException("Family [" + family + "] is missing in the definition."); } if (list.size() - offset != columnNames.size()) { - if (_familyNotInFile) { - if (_autoGenerateRecordIdAsHashOfData) { - throw new IOException("Record [" + str + "] too short, does not match defined record [rowid," - + getColumnNames(columnNames) + "]."); - } else { - throw new IOException("Record [" + str + "] too short, does not match defined record [rowid,recordid," - + getColumnNames(columnNames) + "]."); - } - } else { - if (_autoGenerateRecordIdAsHashOfData) { - throw new IOException("Record [" + str + "] too short, does not match defined record [rowid,family" - + getColumnNames(columnNames) + "]."); - } else { - throw new IOException("Record [" + str + "] too short, does not match defined record [rowid,recordid,family" - + getColumnNames(columnNames) + "]."); - } + + String options = ""; + + if (!_autoGenerateRowIdAsHashOfData) { + options += "rowid,"; + } + if (!_autoGenerateRecordIdAsHashOfData) { + options += "recordid,"; } + if (!_familyNotInFile) { + options += "family,"; + } + String msg = "Record [" + str + "] does not match defined record [" + options + "," + getColumnNames(columnNames) + + "]."; + throw new IOException(msg); } for (int i = 0; i < columnNames.size(); i++) { @@ -459,7 +457,10 @@ public class CsvBlurMapper extends BaseBlurMapper<Writable, Text> { private String getColumnNames(List<String> columnNames) { StringBuilder builder = new StringBuilder(); for (String c : columnNames) { - builder.append(',').append(c); + if (builder.length() != 0) { + builder.append(','); + } + builder.append(c); } return builder.toString(); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/f564924c/distribution/src/main/scripts/bin/blur ---------------------------------------------------------------------- diff --git a/distribution/src/main/scripts/bin/blur b/distribution/src/main/scripts/bin/blur index aa4cd2c..8bed999 100755 --- a/distribution/src/main/scripts/bin/blur +++ b/distribution/src/main/scripts/bin/blur @@ -26,6 +26,12 @@ elif [ $1 = "shell" ]; then "$JAVA_HOME"/bin/java -Dblur.name=$PROC_NAME -Djava.library.path=$JAVA_LIBRARY_PATH $BLUR_COMMAND -Dblur.logs.dir=$BLUR_LOGS -Dblur.log.file=blur-$USER-$PROC_NAME -Dlog4j.configuration=file://$BLUR_HOME/conf/log4j-command.xml -cp $BLUR_CLASSPATH org.apache.blur.shell.Main ${@:2} elif [ $1 = "execute" ]; then "$JAVA_HOME"/bin/java -Dblur.name=$PROC_NAME -Djava.library.path=$JAVA_LIBRARY_PATH $BLUR_COMMAND -Dblur.logs.dir=$BLUR_LOGS -Dblur.log.file=blur-$USER-$PROC_NAME -Dlog4j.configuration=file://$BLUR_HOME/conf/log4j-command.xml -cp $BLUR_CLASSPATH ${@:2} +elif [ $1 = "csv" ]; then + for f in $BLUR_HOME/lib/*.jar; do + BLUR_BASE_CLASSPATH=${BLUR_BASE_CLASSPATH}:$f; + done + export HADOOP_CLASSPATH=$BLUR_BASE_CLASSPATH + hadoop jar $BLUR_HOME/lib/blur-mapred-*.jar org.apache.blur.mapreduce.lib.CsvBlurDriver ${@:2} else "$JAVA_HOME"/bin/java -Dblur.name=$PROC_NAME -Djava.library.path=$JAVA_LIBRARY_PATH $BLUR_COMMAND -Dblur.logs.dir=$BLUR_LOGS -Dblur.log.file=blur-$USER-$PROC_NAME -Dlog4j.configuration=file://$BLUR_HOME/conf/log4j-command.xml -cp $BLUR_CLASSPATH org.apache.blur.shell.Main shell ${@:1} fi http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/f564924c/distribution/src/main/scripts/bin/blur-config.sh ---------------------------------------------------------------------- diff --git a/distribution/src/main/scripts/bin/blur-config.sh b/distribution/src/main/scripts/bin/blur-config.sh index c7251eb..8e539c0 100755 --- a/distribution/src/main/scripts/bin/blur-config.sh +++ b/distribution/src/main/scripts/bin/blur-config.sh @@ -79,6 +79,10 @@ for f in $HADOOP_HOME/lib/*.jar; do BLUR_CLASSPATH=${BLUR_CLASSPATH}:$f; done +for f in $HADOOP_HOME/lib/jsp-*/*.jar; do + BLUR_CLASSPATH=${BLUR_CLASSPATH}:$f; +done + export BLUR_CLASSPATH HOSTNAME=`hostname`
