Repository: incubator-blur Updated Branches: refs/heads/apache-blur-0.2 44a3c1291 -> bb4387e6c
Fixing a few bugs with the index importer via mapreduce. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/bebe68c7 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/bebe68c7 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/bebe68c7 Branch: refs/heads/apache-blur-0.2 Commit: bebe68c7ce2bd19ac374638b4052727cb3c026fb Parents: 44a3c12 Author: Aaron McCurry <[email protected]> Authored: Thu Feb 20 17:53:58 2014 -0500 Committer: Aaron McCurry <[email protected]> Committed: Thu Feb 20 17:53:58 2014 -0500 ---------------------------------------------------------------------- .../apache/blur/manager/writer/BlurIndex.java | 3 +- .../manager/writer/BlurIndexSimpleWriter.java | 10 +- .../apache/blur/manager/writer/IndexAction.java | 37 ++++ .../blur/manager/writer/IndexImporter.java | 128 +++++------ .../blur/manager/writer/MutatableAction.java | 26 ++- .../blur/mapred/AbstractOutputCommitter.java | 18 +- .../lib/BlurOutputFormatMiniClusterTest.java | 211 +++++++++++++++++++ 7 files changed, 364 insertions(+), 69 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/bebe68c7/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java index 5b6310f..55d36c5 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java @@ -26,7 +26,6 @@ import org.apache.blur.lucene.store.refcounter.DirectoryReferenceFileGC; import org.apache.blur.manager.indexserver.BlurIndexWarmup; import org.apache.blur.server.IndexSearcherClosable; import org.apache.blur.server.ShardContext; -import org.apache.blur.thrift.generated.Row; import org.apache.blur.utils.BlurUtil; import org.apache.lucene.index.IndexReader; import org.apache.lucene.index.IndexReaderContext; @@ -134,6 +133,6 @@ public abstract class BlurIndex { return _shardContext; } - public abstract void process(MutatableAction mutatableAction) throws IOException; + public abstract void process(IndexAction indexAction) throws IOException; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/bebe68c7/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java index 26ecb34..64c887b 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java @@ -142,7 +142,7 @@ public class BlurIndexSimpleWriter extends BlurIndex { synchronized (_writer) { _writer.notify(); } - _indexImporter = new IndexImporter(_writer.get(), _lock, _shardContext, TimeUnit.SECONDS, 10); + _indexImporter = new IndexImporter(BlurIndexSimpleWriter.this , _shardContext, TimeUnit.SECONDS, 10); } catch (IOException e) { LOG.error("Unknown error on index writer open.", e); } @@ -294,18 +294,22 @@ public class BlurIndexSimpleWriter extends BlurIndex { } @Override - public void process(MutatableAction mutatableAction) throws IOException { + public void process(IndexAction indexAction) throws IOException { _writeLock.lock(); waitUntilNotNull(_writer); BlurIndexWriter writer = _writer.get(); IndexSearcherClosable indexSearcher = null; try { indexSearcher = getIndexSearcher(); - mutatableAction.performMutate(indexSearcher, writer); + indexAction.performMutate(indexSearcher, writer); + indexAction.doPreCommit(indexSearcher, writer); commit(); + indexAction.doPostCommit(writer); } catch (Exception e) { + indexAction.doPreRollback(writer); writer.rollback(); openWriter(); + indexAction.doPostRollback(writer); throw new IOException("Unknown error during mutation", e); } finally { if (indexSearcher != null) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/bebe68c7/blur-core/src/main/java/org/apache/blur/manager/writer/IndexAction.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/IndexAction.java b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexAction.java new file mode 100644 index 0000000..8fc9272 --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexAction.java @@ -0,0 +1,37 @@ +/** + * 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.manager.writer; + +import java.io.IOException; + +import org.apache.blur.server.IndexSearcherClosable; +import org.apache.lucene.index.BlurIndexWriter; +import org.apache.lucene.index.IndexWriter; + +public abstract class IndexAction { + + public abstract void doPreCommit(IndexSearcherClosable indexSearcher, BlurIndexWriter writer) throws IOException; + + public abstract void doPostCommit(BlurIndexWriter writer) throws IOException; + + public abstract void doPreRollback(BlurIndexWriter writer) throws IOException; + + public abstract void doPostRollback(BlurIndexWriter writer) throws IOException; + + public abstract void performMutate(IndexSearcherClosable searcher, IndexWriter writer) throws IOException; + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/bebe68c7/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java index 30adf17..29b5332 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java @@ -26,11 +26,11 @@ import java.util.Timer; import java.util.TimerTask; import java.util.TreeSet; import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.ReadWriteLock; import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.blur.manager.BlurPartitioner; +import org.apache.blur.server.IndexSearcherClosable; import org.apache.blur.server.ShardContext; import org.apache.blur.store.hdfs.HdfsDirectory; import org.apache.blur.utils.BlurConstants; @@ -43,6 +43,7 @@ import org.apache.hadoop.fs.PathFilter; import org.apache.hadoop.io.Text; import org.apache.lucene.index.AtomicReader; import org.apache.lucene.index.AtomicReaderContext; +import org.apache.lucene.index.BlurIndexWriter; import org.apache.lucene.index.CompositeReaderContext; import org.apache.lucene.index.DirectoryReader; import org.apache.lucene.index.Fields; @@ -56,20 +57,21 @@ import org.apache.lucene.util.BytesRef; public class IndexImporter extends TimerTask implements Closeable { private final static Log LOG = LogFactory.getLog(IndexImporter.class); - private final IndexWriter _indexWriter; - private final ReadWriteLock _lock; + private final BlurIndex _blurIndex; private final ShardContext _shardContext; private final Timer _timer; + private final String _table; + private final String _shard; - public IndexImporter(IndexWriter indexWriter, ReadWriteLock lock, ShardContext shardContext, - TimeUnit refreshUnit, long refreshAmount) { - _indexWriter = indexWriter; - _lock = lock; + public IndexImporter(BlurIndex blurIndex, ShardContext shardContext, TimeUnit refreshUnit, long refreshAmount) { + _blurIndex = blurIndex; _shardContext = shardContext; _timer = new Timer("IndexImporter [" + shardContext.getShard() + "/" + shardContext.getTableContext().getTable() + "]", true); long period = refreshUnit.toMillis(refreshAmount); _timer.schedule(this, period, period); + _table = _shardContext.getTableContext().getTable(); + _shard = _shardContext.getShard(); } @Override @@ -121,61 +123,65 @@ public class IndexImporter extends TimerTask implements Closeable { if (indexesToImport.isEmpty()) { return; } - String table = _shardContext.getTableContext().getTable(); - String shard = _shardContext.getShard(); - for (Directory directory : indexesToImport) { - LOG.info("About to import [{0}] into [{1}/{2}]", directory, shard, table); - } - LOG.info("Obtaining lock on [{0}/{1}]", shard, table); - _lock.writeLock().lock(); - try { + + IndexAction indexAction = getIndexAction(indexesToImport, fileSystem); + _blurIndex.process(indexAction); + } catch (IOException e) { + LOG.error("Unknown error while trying to refresh imports on [{1}/{2}].", e, _shard, _table); + } + } + + private IndexAction getIndexAction(final List<HdfsDirectory> indexesToImport, final FileSystem fileSystem) { + return new IndexAction() { + + private Path _dirPath; + + @Override + public void performMutate(IndexSearcherClosable searcher, IndexWriter writer) throws IOException { + for (Directory directory : indexesToImport) { + LOG.info("About to import [{0}] into [{1}/{2}]", directory, _shard, _table); + } + LOG.info("Obtaining lock on [{0}/{1}]", _shard, _table); for (HdfsDirectory directory : indexesToImport) { - LOG.info("Starting import [{0}], commiting on [{1}/{2}]", directory, shard, table); - _indexWriter.commit(); - boolean isSuccess = true; - boolean isRollbackDueToException = false; - boolean emitDeletes = _indexWriter.numDocs() != 0; - try { - isSuccess = applyDeletes(directory, _indexWriter, shard, emitDeletes); - } catch (IOException e) { - LOG.error("Some issue with deleting the old index on [{0}/{1}]", e, shard, table); - isSuccess = false; - isRollbackDueToException = true; - } - Path dirPath = directory.getPath(); - if (isSuccess) { - LOG.info("Add index [{0}] [{1}/{2}]", directory, shard, table); - _indexWriter.addIndexes(directory); - LOG.info("Removing delete markers [{0}] on [{1}/{2}]", directory, shard, table); - _indexWriter.deleteDocuments(new Term(BlurConstants.DELETE_MARKER, BlurConstants.DELETE_MARKER_VALUE)); - LOG.info("Finishing import [{0}], commiting on [{1}/{2}]", directory, shard, table); - _indexWriter.commit(); - _indexWriter.maybeMerge(); - LOG.info("Cleaning up old directory [{0}] for [{1}/{2}]", dirPath, shard, table); - fileSystem.delete(dirPath, true); - LOG.info("Import complete on [{0}/{1}]", shard, table); - } else { - if (!isRollbackDueToException) { - LOG.error( - "Index is corrupted, RowIds are found in wrong shard [{0}/{1}], cancelling index import for [{2}]", - shard, table, directory); - } - LOG.info("Starting rollback on [{0}/{1}]", shard, table); - _indexWriter.rollback(); - LOG.info("Finished rollback on [{0}/{1}]", shard, table); - String name = dirPath.getName(); - int lastIndexOf = name.lastIndexOf('.'); - String badRowIdsName = name.substring(0, lastIndexOf) + ".bad_rowids"; - fileSystem.rename(dirPath, new Path(dirPath.getParent(), badRowIdsName)); - } + boolean emitDeletes = searcher.getIndexReader().numDocs() != 0; + _dirPath = directory.getPath(); + applyDeletes(directory, writer, _shard, emitDeletes); + LOG.info("Add index [{0}] [{1}/{2}]", directory, _shard, _table); + writer.addIndexes(directory); + LOG.info("Removing delete markers [{0}] on [{1}/{2}]", directory, _shard, _table); + writer.deleteDocuments(new Term(BlurConstants.DELETE_MARKER, BlurConstants.DELETE_MARKER_VALUE)); + LOG.info("Finishing import [{0}], commiting on [{1}/{2}]", directory, _shard, _table); } - } finally { - _lock.writeLock().unlock(); } - } catch (IOException e) { - LOG.error("Unknown error while trying to refresh imports.", e); - } + @Override + public void doPreCommit(IndexSearcherClosable indexSearcher, BlurIndexWriter writer) throws IOException { + + } + + @Override + public void doPostCommit(BlurIndexWriter writer) throws IOException { + LOG.info("Calling maybeMerge on the index [{0}] for [{1}/{2}]", _dirPath, _shard, _table); + writer.maybeMerge(); + LOG.info("Cleaning up old directory [{0}] for [{1}/{2}]", _dirPath, _shard, _table); + fileSystem.delete(_dirPath, true); + LOG.info("Import complete on [{0}/{1}]", _shard, _table); + } + + @Override + public void doPreRollback(BlurIndexWriter writer) throws IOException { + LOG.info("Starting rollback on [{0}/{1}]", _shard, _table); + } + + @Override + public void doPostRollback(BlurIndexWriter writer) throws IOException { + LOG.info("Finished rollback on [{0}/{1}]", _shard, _table); + String name = _dirPath.getName(); + int lastIndexOf = name.lastIndexOf('.'); + String badRowIdsName = name.substring(0, lastIndexOf) + ".bad_rowids"; + fileSystem.rename(_dirPath, new Path(_dirPath.getParent(), badRowIdsName)); + } + }; } private SortedSet<FileStatus> sort(FileStatus[] listStatus) { @@ -186,7 +192,7 @@ public class IndexImporter extends TimerTask implements Closeable { return result; } - private boolean applyDeletes(Directory directory, IndexWriter indexWriter, String shard, boolean emitDeletes) + private void applyDeletes(Directory directory, IndexWriter indexWriter, String shard, boolean emitDeletes) throws IOException { DirectoryReader reader = DirectoryReader.open(directory); try { @@ -208,7 +214,8 @@ public class IndexImporter extends TimerTask implements Closeable { key.set(ref.bytes, ref.offset, ref.length); int partition = blurPartitioner.getPartition(key, null, numberOfShards); if (shardId != partition) { - return false; + throw new IOException("Index is corrupted, RowIds are found in wrong shard, partition [" + partition + + "] does not shard [" + shardId + "], this can happen when rows are not hashed correctly."); } if (emitDeletes) { indexWriter.deleteDocuments(new Term(BlurConstants.ROW_ID, BytesRef.deepCopyOf(ref))); @@ -219,6 +226,5 @@ public class IndexImporter extends TimerTask implements Closeable { } finally { reader.close(); } - return true; } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/bebe68c7/blur-core/src/main/java/org/apache/blur/manager/writer/MutatableAction.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/MutatableAction.java b/blur-core/src/main/java/org/apache/blur/manager/writer/MutatableAction.java index 6d70c86..cbe7ee2 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/MutatableAction.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/MutatableAction.java @@ -42,6 +42,7 @@ import org.apache.blur.thrift.generated.Selector; import org.apache.blur.utils.BlurConstants; import org.apache.blur.utils.RowDocumentUtil; import org.apache.lucene.document.Field; +import org.apache.lucene.index.BlurIndexWriter; import org.apache.lucene.index.IndexWriter; import org.apache.lucene.index.Term; @@ -49,7 +50,7 @@ import com.yammer.metrics.Metrics; import com.yammer.metrics.core.Meter; import com.yammer.metrics.core.MetricName; -public class MutatableAction { +public class MutatableAction extends IndexAction { private static final Meter _writeRecordsMeter; private static final Meter _writeRowMeter; @@ -327,7 +328,8 @@ public class MutatableAction { updateRow.replaceColumns(record); } - void performMutate(IndexSearcherClosable searcher, IndexWriter writer) throws IOException { + @Override + public void performMutate(IndexSearcherClosable searcher, IndexWriter writer) throws IOException { try { for (InternalAction internalAction : _actions) { internalAction.performAction(searcher, writer); @@ -355,4 +357,24 @@ public class MutatableAction { return updateRow; } + @Override + public void doPreCommit(IndexSearcherClosable indexSearcher, BlurIndexWriter writer) { + + } + + @Override + public void doPostCommit(BlurIndexWriter writer) { + + } + + @Override + public void doPreRollback(BlurIndexWriter writer) { + + } + + @Override + public void doPostRollback(BlurIndexWriter writer) { + + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/bebe68c7/blur-mapred/src/main/java/org/apache/blur/mapred/AbstractOutputCommitter.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapred/AbstractOutputCommitter.java b/blur-mapred/src/main/java/org/apache/blur/mapred/AbstractOutputCommitter.java index 4f110e2..8294738 100644 --- a/blur-mapred/src/main/java/org/apache/blur/mapred/AbstractOutputCommitter.java +++ b/blur-mapred/src/main/java/org/apache/blur/mapred/AbstractOutputCommitter.java @@ -28,6 +28,7 @@ import org.apache.hadoop.conf.Configuration; 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.mapred.JobContext; import org.apache.hadoop.mapred.OutputCommitter; import org.apache.hadoop.mapred.TaskAttemptID; @@ -71,13 +72,25 @@ public abstract class AbstractOutputCommitter extends OutputCommitter { private void commitOrAbortJob(JobContext jobContext, Path shardPath, boolean commit) throws IOException { FileSystem fileSystem = shardPath.getFileSystem(jobContext.getConfiguration()); - FileStatus[] listStatus = fileSystem.listStatus(shardPath); + FileStatus[] listStatus = fileSystem.listStatus(shardPath, new PathFilter() { + @Override + public boolean accept(Path path) { + if (path.getName().endsWith(".task_complete")) { + return true; + } + return false; + } + }); for (FileStatus fileStatus : listStatus) { Path path = fileStatus.getPath(); String name = path.getName(); boolean taskComplete = name.endsWith(".task_complete"); if (fileStatus.isDir()) { String taskAttemptName = getTaskAttemptName(name); + if (taskAttemptName == null) { + LOG.info("Dir name [{0}] not task attempt", name); + continue; + } TaskAttemptID taskAttemptID = TaskAttemptID.forName(taskAttemptName); if (taskAttemptID.getJobID().equals(jobContext.getJobID())) { if (commit) { @@ -99,6 +112,9 @@ public abstract class AbstractOutputCommitter extends OutputCommitter { private String getTaskAttemptName(String name) { int lastIndexOf = name.lastIndexOf('.'); + if (lastIndexOf < 0) { + return null; + } return name.substring(0, lastIndexOf); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/bebe68c7/blur-mapred/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatMiniClusterTest.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatMiniClusterTest.java b/blur-mapred/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatMiniClusterTest.java new file mode 100644 index 0000000..f46a6ba --- /dev/null +++ b/blur-mapred/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatMiniClusterTest.java @@ -0,0 +1,211 @@ +package org.apache.blur.mapreduce.lib; + +/** + * 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. + */ +import static org.junit.Assert.assertTrue; + +import java.io.BufferedReader; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.PrintWriter; + +import org.apache.blur.MiniCluster; +import org.apache.blur.server.TableContext; +import org.apache.blur.store.buffer.BufferStore; +import org.apache.blur.thirdparty.thrift_0_9_0.TException; +import org.apache.blur.thrift.BlurClient; +import org.apache.blur.thrift.generated.Blur.Iface; +import org.apache.blur.thrift.generated.BlurException; +import org.apache.blur.thrift.generated.TableDescriptor; +import org.apache.blur.thrift.generated.TableStats; +import org.apache.blur.utils.GCWatcher; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.LocalFileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.permission.FsAction; +import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.MiniMRCluster; +import org.apache.hadoop.mapreduce.Counters; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.TestMapReduceLocal.TrackingTextInputFormat; +import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +public class BlurOutputFormatMiniClusterTest { + + private static Configuration conf = new Configuration(); + private static FileSystem fileSystem; + private static MiniMRCluster mr; + private static Path TEST_ROOT_DIR; + private static JobConf jobConf; + private static MiniCluster miniCluster; + private Path inDir = new Path(TEST_ROOT_DIR + "/in"); + private static final File TMPDIR = new File(System.getProperty("blur.tmp.dir", + "./target/tmp_BlurOutputFormatMiniClusterTest")); + + @BeforeClass + public static void setupTest() throws Exception { + GCWatcher.init(0.60); + LocalFileSystem localFS = FileSystem.getLocal(new Configuration()); + File testDirectory = new File(TMPDIR, "blur-cluster-test").getAbsoluteFile(); + testDirectory.mkdirs(); + + Path directory = new Path(testDirectory.getPath()); + FsPermission dirPermissions = localFS.getFileStatus(directory).getPermission(); + FsAction userAction = dirPermissions.getUserAction(); + FsAction groupAction = dirPermissions.getGroupAction(); + FsAction otherAction = dirPermissions.getOtherAction(); + + StringBuilder builder = new StringBuilder(); + builder.append(userAction.ordinal()); + builder.append(groupAction.ordinal()); + builder.append(otherAction.ordinal()); + String dirPermissionNum = builder.toString(); + System.setProperty("dfs.datanode.data.dir.perm", dirPermissionNum); + testDirectory.delete(); + miniCluster = new MiniCluster(); + miniCluster.startBlurCluster(new File(testDirectory, "cluster").getAbsolutePath(), 2, 3, true); + + // System.setProperty("test.build.data", + // "./target/BlurOutputFormatTest/data"); + // TEST_ROOT_DIR = new Path(System.getProperty("test.build.data", + // "target/tmp/BlurOutputFormatTest_tmp")); + TEST_ROOT_DIR = new Path(miniCluster.getFileSystemUri().toString() + "/blur_test"); + System.setProperty("hadoop.log.dir", "./target/BlurOutputFormatTest/hadoop_log"); + try { + fileSystem = TEST_ROOT_DIR.getFileSystem(conf); + } catch (IOException io) { + throw new RuntimeException("problem getting local fs", io); + } + mr = new MiniMRCluster(1, miniCluster.getFileSystemUri().toString(), 1); + jobConf = mr.createJobConf(); + BufferStore.initNewBuffer(128, 128 * 128); + } + + @AfterClass + public static void teardown() { + if (mr != null) { + mr.shutdown(); + } + miniCluster.shutdownBlurCluster(); + rm(new File("build")); + } + + private static void rm(File file) { + if (!file.exists()) { + return; + } + if (file.isDirectory()) { + for (File f : file.listFiles()) { + rm(f); + } + } + file.delete(); + } + + @Before + public void setup() { + TableContext.clear(); + } + + @Test + public void testBlurOutputFormat() throws IOException, InterruptedException, ClassNotFoundException, BlurException, + TException { + fileSystem.delete(inDir, true); + String tableName = "testBlurOutputFormat"; + writeRecordsFile("in/part1", 1, 1, 1, 1, "cf1"); + writeRecordsFile("in/part2", 1, 1, 2, 1, "cf1"); + + Job job = new Job(jobConf, "blur index"); + job.setJarByClass(BlurOutputFormatMiniClusterTest.class); + job.setMapperClass(CsvBlurMapper.class); + job.setInputFormatClass(TrackingTextInputFormat.class); + + FileInputFormat.addInputPath(job, new Path(TEST_ROOT_DIR + "/in")); + String tableUri = new Path(TEST_ROOT_DIR + "/blur/" + tableName).toString(); + CsvBlurMapper.addColumns(job, "cf1", "col"); + + TableDescriptor tableDescriptor = new TableDescriptor(); + tableDescriptor.setShardCount(1); + tableDescriptor.setTableUri(tableUri); + tableDescriptor.setName(tableName); + + Iface client = getClient(); + client.createTable(tableDescriptor); + + BlurOutputFormat.setupJob(job, tableDescriptor); + + assertTrue(job.waitForCompletion(true)); + Counters ctrs = job.getCounters(); + System.out.println("Counters: " + ctrs); + + while (true) { + TableStats tableStats = client.tableStats(tableName); + System.out.println(tableStats); + if (tableStats.getRowCount() > 0) { + break; + } + Thread.sleep(5000); + } + } + + private Iface getClient() { + return BlurClient.getClient(miniCluster.getControllerConnectionStr()); + } + + public static String readFile(String name) throws IOException { + DataInputStream f = fileSystem.open(new Path(TEST_ROOT_DIR + "/" + name)); + BufferedReader b = new BufferedReader(new InputStreamReader(f)); + StringBuilder result = new StringBuilder(); + String line = b.readLine(); + while (line != null) { + result.append(line); + result.append('\n'); + line = b.readLine(); + } + b.close(); + return result.toString(); + } + + private Path writeRecordsFile(String name, int starintgRowId, int numberOfRows, int startRecordId, + int numberOfRecords, String family) throws IOException { + // "1,1,cf1,val1" + Path file = new Path(TEST_ROOT_DIR + "/" + name); + fileSystem.delete(file, false); + DataOutputStream f = fileSystem.create(file); + PrintWriter writer = new PrintWriter(f); + for (int row = 0; row < numberOfRows; row++) { + for (int record = 0; record < numberOfRecords; record++) { + writer.println(getRecord(row + starintgRowId, record + startRecordId, family)); + } + } + writer.close(); + return file; + } + + private String getRecord(int rowId, int recordId, String family) { + return rowId + "," + recordId + "," + family + ",valuetoindex"; + } +}
