Making some performance updates to the blur record writer used in MR indexing.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/928b3405 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/928b3405 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/928b3405 Branch: refs/heads/apache-blur-0.2 Commit: 928b3405df31a7365d2d63ab30a513dcb96de6d5 Parents: 2286f25 Author: Aaron McCurry <[email protected]> Authored: Tue Jun 17 10:23:42 2014 -0400 Committer: Aaron McCurry <[email protected]> Committed: Tue Jun 17 10:23:42 2014 -0400 ---------------------------------------------------------------------- .../mapreduce/lib/GenericBlurRecordWriter.java | 69 +++++++++++++------- .../apache/blur/mapreduce/lib/NullCounter.java | 24 +++++++ .../mapreduce/lib/GenericBlurRecordWriter.java | 32 ++++++--- 3 files changed, 92 insertions(+), 33 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/928b3405/blur-mapred-hadoop1/src/main/java/org/apache/blur/mapreduce/lib/GenericBlurRecordWriter.java ---------------------------------------------------------------------- diff --git a/blur-mapred-hadoop1/src/main/java/org/apache/blur/mapreduce/lib/GenericBlurRecordWriter.java b/blur-mapred-hadoop1/src/main/java/org/apache/blur/mapreduce/lib/GenericBlurRecordWriter.java index e138cd5..0ec7dd3 100644 --- a/blur-mapred-hadoop1/src/main/java/org/apache/blur/mapreduce/lib/GenericBlurRecordWriter.java +++ b/blur-mapred-hadoop1/src/main/java/org/apache/blur/mapreduce/lib/GenericBlurRecordWriter.java @@ -23,6 +23,7 @@ import java.util.List; import java.util.Map; import java.util.TreeMap; import java.util.UUID; +import java.util.concurrent.TimeUnit; import org.apache.blur.analysis.FieldManager; import org.apache.blur.log.Log; @@ -52,17 +53,17 @@ import org.apache.lucene.index.CorruptIndexException; import org.apache.lucene.index.DirectoryReader; import org.apache.lucene.index.IndexWriter; import org.apache.lucene.index.IndexWriterConfig; -import org.apache.lucene.index.NoMergePolicy; import org.apache.lucene.index.TieredMergePolicy; import org.apache.lucene.store.Directory; -import org.apache.lucene.store.FSDirectory; import org.apache.lucene.store.IOContext; import org.apache.lucene.store.NoLockFactory; +import org.apache.lucene.store.SimpleFSDirectory; public class GenericBlurRecordWriter { private static final Log LOG = LogFactory.getLog(GenericBlurRecordWriter.class); private static final String JAVA_IO_TMPDIR = "java.io.tmpdir"; + private static final Counter NULL_COUNTER = new NullCounter(); private final Text _prevKey = new Text(); private final Map<String, List<Field>> _documents = new TreeMap<String, List<Field>>(); @@ -77,38 +78,39 @@ public class GenericBlurRecordWriter { private final Path _newIndex; private final boolean _indexLocally; private final boolean _optimizeInFlight; - private Counter _columnCount; - private Counter _fieldCount; - private Counter _recordCount; - private Counter _rowCount; - private Counter _recordDuplicateCount; - private Counter _rowOverFlowCount; - private Counter _rowDeleteCount; - private RateCounter _recordRateCounter; - private RateCounter _rowRateCounter; - private RateCounter _copyRateCounter; + private Counter _columnCount = NULL_COUNTER; + private Counter _fieldCount = NULL_COUNTER; + private Counter _recordCount = NULL_COUNTER; + private Counter _rowCount = NULL_COUNTER; + private Counter _recordDuplicateCount = NULL_COUNTER; + private Counter _rowOverFlowCount = NULL_COUNTER; + private Counter _rowDeleteCount = NULL_COUNTER; + private RateCounter _recordRateCounter = new RateCounter(NULL_COUNTER); + private RateCounter _rowRateCounter = new RateCounter(NULL_COUNTER); + private RateCounter _copyRateCounter = new RateCounter(NULL_COUNTER); private boolean _countersSetup = false; private IndexWriter _localTmpWriter; private boolean _usingLocalTmpindex; private File _localTmpPath; private ProgressableDirectory _localTmpDir; private String _deletedRowId; + private Configuration _configuration; public GenericBlurRecordWriter(Configuration configuration, int attemptId, String tmpDirName) throws IOException { + _configuration = configuration; + _indexLocally = BlurOutputFormat.isIndexLocally(_configuration); + _optimizeInFlight = BlurOutputFormat.isOptimizeInFlight(_configuration); - _indexLocally = BlurOutputFormat.isIndexLocally(configuration); - _optimizeInFlight = BlurOutputFormat.isOptimizeInFlight(configuration); - - TableDescriptor tableDescriptor = BlurOutputFormat.getTableDescriptor(configuration); + TableDescriptor tableDescriptor = BlurOutputFormat.getTableDescriptor(_configuration); int shardCount = tableDescriptor.getShardCount(); int shardId = attemptId % shardCount; - _maxDocumentBufferSize = BlurOutputFormat.getMaxDocumentBufferSize(configuration); - Path tableOutput = BlurOutputFormat.getOutputPath(configuration); + _maxDocumentBufferSize = BlurOutputFormat.getMaxDocumentBufferSize(_configuration); + Path tableOutput = BlurOutputFormat.getOutputPath(_configuration); String shardName = BlurUtil.getShardName(BlurConstants.SHARD_PREFIX, shardId); Path indexPath = new Path(tableOutput, shardName); _newIndex = new Path(indexPath, tmpDirName); - _finalDir = new ProgressableDirectory(new HdfsDirectory(configuration, _newIndex), getProgressable()); + _finalDir = new ProgressableDirectory(new HdfsDirectory(_configuration, _newIndex), getProgressable()); _finalDir.setLockFactory(NoLockFactory.getNoLockFactory()); TableContext tableContext = TableContext.create(tableDescriptor); @@ -122,12 +124,12 @@ public class GenericBlurRecordWriter { mergePolicy.setUseCompoundFile(false); _overFlowConf = _conf.clone(); - _overFlowConf.setMergePolicy(NoMergePolicy.NO_COMPOUND_FILES); if (_indexLocally) { String localDirPath = System.getProperty(JAVA_IO_TMPDIR); _localPath = new File(localDirPath, UUID.randomUUID().toString() + ".tmp"); - _localDir = new ProgressableDirectory(FSDirectory.open(_localPath), getProgressable()); + SimpleFSDirectory directory = new SimpleFSDirectory(_localPath); + _localDir = new ProgressableDirectory(directory, getProgressable()); _writer = new IndexWriter(_localDir, _conf.clone()); } else { _localPath = null; @@ -137,12 +139,27 @@ public class GenericBlurRecordWriter { } private Progressable getProgressable() { + final Progressable prg = BlurOutputFormat.getProgressable(); return new Progressable() { + + private Progressable _progressable = prg; + private long _lastWarn = 0; + @Override public void progress() { - Progressable progressable = BlurOutputFormat.getProgressable(); - if (progressable != null) { - progressable.progress(); + if (_progressable != null) { + _progressable.progress(); + } else { + Progressable progressable = BlurOutputFormat.getProgressable(); + if (progressable != null) { + _progressable = progressable; + } else { + long now = System.nanoTime(); + if (_lastWarn + TimeUnit.SECONDS.toNanos(10) < now) { + LOG.warn("Progress not being reported."); + _lastWarn = System.nanoTime(); + } + } } } }; @@ -212,7 +229,8 @@ public class GenericBlurRecordWriter { if (_localTmpWriter == null) { String localDirPath = System.getProperty(JAVA_IO_TMPDIR); _localTmpPath = new File(localDirPath, UUID.randomUUID().toString() + ".tmp"); - _localTmpDir = new ProgressableDirectory(FSDirectory.open(_localTmpPath), BlurOutputFormat.getProgressable()); + SimpleFSDirectory directory = new SimpleFSDirectory(_localTmpPath); + _localTmpDir = new ProgressableDirectory(directory, 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. @@ -257,6 +275,7 @@ public class GenericBlurRecordWriter { _writer.addIndexes(reader); reader.close(); resetLocalTmp(); + _writer.maybeMerge(); if (_countersSetup) { _rowOverFlowCount.increment(1); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/928b3405/blur-mapred-hadoop1/src/main/java/org/apache/blur/mapreduce/lib/NullCounter.java ---------------------------------------------------------------------- diff --git a/blur-mapred-hadoop1/src/main/java/org/apache/blur/mapreduce/lib/NullCounter.java b/blur-mapred-hadoop1/src/main/java/org/apache/blur/mapreduce/lib/NullCounter.java new file mode 100644 index 0000000..b13f55f --- /dev/null +++ b/blur-mapred-hadoop1/src/main/java/org/apache/blur/mapreduce/lib/NullCounter.java @@ -0,0 +1,24 @@ +package org.apache.blur.mapreduce.lib; + +import org.apache.hadoop.mapreduce.Counter; + +/** + * 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. + */ + +public class NullCounter extends Counter { + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/928b3405/blur-mapred-hadoop2/src/main/java/org/apache/blur/mapreduce/lib/GenericBlurRecordWriter.java ---------------------------------------------------------------------- diff --git a/blur-mapred-hadoop2/src/main/java/org/apache/blur/mapreduce/lib/GenericBlurRecordWriter.java b/blur-mapred-hadoop2/src/main/java/org/apache/blur/mapreduce/lib/GenericBlurRecordWriter.java index bda61be..0ec7dd3 100644 --- a/blur-mapred-hadoop2/src/main/java/org/apache/blur/mapreduce/lib/GenericBlurRecordWriter.java +++ b/blur-mapred-hadoop2/src/main/java/org/apache/blur/mapreduce/lib/GenericBlurRecordWriter.java @@ -23,6 +23,7 @@ import java.util.List; import java.util.Map; import java.util.TreeMap; import java.util.UUID; +import java.util.concurrent.TimeUnit; import org.apache.blur.analysis.FieldManager; import org.apache.blur.log.Log; @@ -52,11 +53,11 @@ import org.apache.lucene.index.CorruptIndexException; import org.apache.lucene.index.DirectoryReader; import org.apache.lucene.index.IndexWriter; import org.apache.lucene.index.IndexWriterConfig; -import org.apache.lucene.index.NoMergePolicy; import org.apache.lucene.index.TieredMergePolicy; import org.apache.lucene.store.Directory; import org.apache.lucene.store.IOContext; import org.apache.lucene.store.NoLockFactory; +import org.apache.lucene.store.SimpleFSDirectory; public class GenericBlurRecordWriter { @@ -123,12 +124,11 @@ public class GenericBlurRecordWriter { mergePolicy.setUseCompoundFile(false); _overFlowConf = _conf.clone(); - _overFlowConf.setMergePolicy(NoMergePolicy.NO_COMPOUND_FILES); if (_indexLocally) { String localDirPath = System.getProperty(JAVA_IO_TMPDIR); _localPath = new File(localDirPath, UUID.randomUUID().toString() + ".tmp"); - HdfsDirectory directory = new HdfsDirectory(_configuration, new Path(_localPath.toURI())); + SimpleFSDirectory directory = new SimpleFSDirectory(_localPath); _localDir = new ProgressableDirectory(directory, getProgressable()); _writer = new IndexWriter(_localDir, _conf.clone()); } else { @@ -139,12 +139,27 @@ public class GenericBlurRecordWriter { } private Progressable getProgressable() { + final Progressable prg = BlurOutputFormat.getProgressable(); return new Progressable() { + + private Progressable _progressable = prg; + private long _lastWarn = 0; + @Override public void progress() { - Progressable progressable = BlurOutputFormat.getProgressable(); - if (progressable != null) { - progressable.progress(); + if (_progressable != null) { + _progressable.progress(); + } else { + Progressable progressable = BlurOutputFormat.getProgressable(); + if (progressable != null) { + _progressable = progressable; + } else { + long now = System.nanoTime(); + if (_lastWarn + TimeUnit.SECONDS.toNanos(10) < now) { + LOG.warn("Progress not being reported."); + _lastWarn = System.nanoTime(); + } + } } } }; @@ -214,8 +229,8 @@ public class GenericBlurRecordWriter { if (_localTmpWriter == null) { String localDirPath = System.getProperty(JAVA_IO_TMPDIR); _localTmpPath = new File(localDirPath, UUID.randomUUID().toString() + ".tmp"); - HdfsDirectory directory = new HdfsDirectory(_configuration, new Path(_localTmpPath.toURI())); - _localTmpDir = new ProgressableDirectory(directory, BlurOutputFormat.getProgressable()); + SimpleFSDirectory directory = new SimpleFSDirectory(_localTmpPath); + _localTmpDir = new ProgressableDirectory(directory, 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. @@ -260,6 +275,7 @@ public class GenericBlurRecordWriter { _writer.addIndexes(reader); reader.close(); resetLocalTmp(); + _writer.maybeMerge(); if (_countersSetup) { _rowOverFlowCount.increment(1); }
