accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ujustgotbi...@apache.org
Subject [18/50] git commit: ACCUMULO-375 hybridized ingest to use some bulk and some streaming
Date Thu, 06 Feb 2014 05:39:57 GMT
ACCUMULO-375 hybridized ingest to use some bulk and some streaming

git-svn-id: https://svn.apache.org/repos/asf/incubator/accumulo/branches/1.4@1245142 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/ec56d2d4
Tree: http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/tree/ec56d2d4
Diff: http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/diff/ec56d2d4

Branch: refs/heads/1.4.5-SNAPSHOT
Commit: ec56d2d429ebf4bd849daf71804021536f4d21ee
Parents: 0e1e67d
Author: Adam Fuchs <afuchs@apache.org>
Authored: Thu Feb 16 19:54:31 2012 +0000
Committer: Adam Fuchs <afuchs@apache.org>
Committed: Thu Feb 16 19:54:31 2012 +0000

----------------------------------------------------------------------
 .../ingest/WikipediaPartitionedMapper.java        | 18 +++++++++++++++---
 .../output/BufferingRFileRecordWriter.java        |  9 +++++----
 2 files changed, 20 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/blob/ec56d2d4/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
index 25bf572..7816b03 100644
--- 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
@@ -26,6 +26,9 @@ import java.util.HashSet;
 import java.util.Map.Entry;
 import java.util.Set;
 
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.MultiTableBatchWriter;
 import org.apache.accumulo.core.data.Mutation;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.security.ColumnVisibility;
@@ -112,6 +115,7 @@ public class WikipediaPartitionedMapper extends Mapper<Text,Article,Text,Mutatio
     }
   }
   
+  MultiTableBatchWriter mtbw;
 
   @Override
   public void setup(final Context context) {
@@ -121,6 +125,14 @@ public class WikipediaPartitionedMapper extends Mapper<Text,Article,Text,Mutatio
     reverseIndexTableName = new Text(tablename + "ReverseIndex");
     metadataTableName = new Text(tablename + "Metadata");
     
+    try {
+      mtbw = WikipediaConfiguration.getConnector(conf).createMultiTableBatchWriter(10000000,
1000, 10);
+    } catch (AccumuloException e) {
+      throw new RuntimeException(e);
+    } catch (AccumuloSecurityException e) {
+      throw new RuntimeException(e);
+    }
+    
     final Text metadataTableNameFinal = metadataTableName;
     final Text indexTableNameFinal = indexTableName;
     final Text reverseIndexTableNameFinal = reverseIndexTableName;
@@ -163,7 +175,7 @@ public class WikipediaPartitionedMapper extends Mapper<Text,Article,Text,Mutatio
           Mutation m = new Mutation(key.row);
           m.put(key.colfam, key.colqual, key.cv, key.timestamp, val);
           try {
-            context.write(indexTableNameFinal, m);
+            mtbw.getBatchWriter(indexTableNameFinal.toString()).addMutation(m);
           } catch (Exception e) {
             throw new RuntimeException(e);
           }
@@ -189,7 +201,7 @@ public class WikipediaPartitionedMapper extends Mapper<Text,Article,Text,Mutatio
           Mutation m = new Mutation(key.row);
           m.put(key.colfam, key.colqual, key.cv, key.timestamp, val);
           try {
-            context.write(reverseIndexTableNameFinal, m);
+            mtbw.getBatchWriter(reverseIndexTableNameFinal.toString()).addMutation(m);
           } catch (Exception e) {
             throw new RuntimeException(e);
           }
@@ -210,7 +222,7 @@ public class WikipediaPartitionedMapper extends Mapper<Text,Article,Text,Mutatio
             Mutation m = new Mutation(key.row);
             m.put(key.colfam, key.colqual, key.cv, key.timestamp, value);
             try {
-              context.write(metadataTableNameFinal, m);
+              mtbw.getBatchWriter(metadataTableNameFinal.toString()).addMutation(m);
             } catch (Exception e) {
               throw new RuntimeException(e);
             }

http://git-wip-us.apache.org/repos/asf/accumulo-wikisearch/blob/ec56d2d4/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/output/BufferingRFileRecordWriter.java
----------------------------------------------------------------------
diff --git a/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/output/BufferingRFileRecordWriter.java
b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/output/BufferingRFileRecordWriter.java
index a7e7dcf..579bbe1 100644
--- a/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/output/BufferingRFileRecordWriter.java
+++ b/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/output/BufferingRFileRecordWriter.java
@@ -69,8 +69,8 @@ final class BufferingRFileRecordWriter extends RecordWriter<Text,Mutation>
{
     if (buffer.size() == 0)
       return;
     
-    // TODO fix the filename
     String file = filenamePrefix + "/" + tablename + "/" + taskID + "_" + (fileCount++) +
".rf";
+    // TODO get the table configuration for the given table?
     FileSKVWriter writer = RFileOperations.getInstance().openWriter(file, fs, conf, acuconf);
     
     // forget locality groups for now, just write everything to the default
@@ -110,17 +110,18 @@ final class BufferingRFileRecordWriter extends RecordWriter<Text,Mutation>
{
     {
       Key k = new Key(mutation.getRow(),update.getColumnFamily(),update.getColumnQualifier(),update.getColumnVisibility(),update.getTimestamp(),update.isDeleted());
       Value v = new Value(update.getValue());
+      // TODO account for object overhead
       mutationSize += k.getSize();
       mutationSize += v.getSize();
       buffer.put(k, v);
     }
     size += mutationSize;
     long bufferSize = bufferSizes.get(table);
+    
+    // TODO use a MutableLong instead
     bufferSize += mutationSize;
     bufferSizes.put(table, bufferSize);
-    
-    // TODO add object overhead size
-    
+
     while (size >= maxSize) {
       flushLargestTable();
     }


Mime
View raw message