You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by li...@apache.org on 2013/01/13 20:21:09 UTC
svn commit: r1432708 - in /hbase/branches/0.89-fb/src:
main/java/org/apache/hadoop/hbase/client/
main/java/org/apache/hadoop/hbase/mapreduce/
main/java/org/apache/hadoop/hbase/regionserver/
test/java/org/apache/hadoop/hbase/mapreduce/
Author: liyin
Date: Sun Jan 13 19:21:08 2013
New Revision: 1432708
URL: http://svn.apache.org/viewvc?rev=1432708&view=rev
Log:
[HBASE-7424] Enable the DeltaEncoding for the HFileOutputFormat
Author: manukranthk
Summary: HFileOutputFormat has a writer embedded but it is not configured to use the DeltaEncoding and FavoredNodes. This revision is to add that support to the HFileOutputFormat while it is used as an OutputFormat either the Mapper or the Reducer for a MapReduce task.
Test Plan: Adding Unit Tests to test the working of the HFileOutputFormat
Reviewers: liyintang, kannan
CC: hbase-eng@, astewart
Differential Revision: https://phabricator.fb.com/D669389
Task ID: 1964673
Modified:
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/client/HTable.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/mapreduce/HFileOutputFormat.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java
hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/mapreduce/TestHFileOutputFormat.java
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/client/HTable.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/client/HTable.java?rev=1432708&r1=1432707&r2=1432708&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/client/HTable.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/client/HTable.java Sun Jan 13 19:21:08 2013
@@ -378,7 +378,7 @@ public class HTable implements HTableInt
return new Pair(startKeyList.toArray(new byte[startKeyList.size()][]),
endKeyList.toArray(new byte[endKeyList.size()][]));
}
-
+
/**
* Gets all the regions and their address for this table.
* <p>
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/mapreduce/HFileOutputFormat.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/mapreduce/HFileOutputFormat.java?rev=1432708&r1=1432707&r2=1432708&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/mapreduce/HFileOutputFormat.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/mapreduce/HFileOutputFormat.java Sun Jan 13 19:21:08 2013
@@ -21,6 +21,7 @@ package org.apache.hadoop.hbase.mapreduc
import java.io.IOException;
import java.io.UnsupportedEncodingException;
+import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URLDecoder;
@@ -46,14 +47,18 @@ import org.apache.hadoop.hbase.client.HT
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.RowMutation;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.io.hfile.Compression;
import org.apache.hadoop.hbase.io.hfile.HFile;
+import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoder;
+import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoderImpl;
import org.apache.hadoop.hbase.mapreduce.hadoopbackport.TotalOrderPartitioner;
import org.apache.hadoop.hbase.regionserver.StoreFile.BloomType;
import org.apache.hadoop.hbase.regionserver.StoreFile;
import org.apache.hadoop.hbase.util.BloomFilterFactory;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.mapreduce.Job;
@@ -76,6 +81,11 @@ public class HFileOutputFormat extends F
//This stores a string in the format family1=bloomType1&family2=bloomType2&...&familyN=bloomTypeN
static final String BLOOMFILTER_TYPE_PER_CF_KEY =
"hbase.hfileoutputformat.families.bloomfilter.typePerCF";
+
+ static final String ENCODING_TYPE_PER_CF_KEY =
+ "hbase.hfileoutputformat.families.encoding.typePerCF";
+
+ static final String UTF8 = "UTF-8";
public RecordWriter<ImmutableBytesWritable, KeyValue> getRecordWriter(final TaskAttemptContext context)
throws IOException, InterruptedException {
@@ -97,6 +107,9 @@ public class HFileOutputFormat extends F
final Map<byte[], String> compressionMap = createFamilyCompressionMap(conf);
final Map<byte[], BloomType> bloomTypeMap = createFamilyBloomTypeMap(conf);
+
+ final Map<byte[], HFileDataBlockEncoder> encoderTypeMap =
+ createFamilyDeltaEncodingMap(conf);
return new RecordWriter<ImmutableBytesWritable, KeyValue>() {
// Map of families to writers and how much has been output on the writer.
@@ -175,10 +188,17 @@ public class HFileOutputFormat extends F
String compression = compressionMap.get(family);
compression = compression == null ? defaultCompression : compression;
BloomType bloomType = bloomTypeMap.get(family);
+ HFileDataBlockEncoder encoder = encoderTypeMap.get(family);
if (bloomType == null) {
bloomType = BloomType.NONE;
}
-
+ if (encoder == null) {
+ encoder = new HFileDataBlockEncoderImpl(DataBlockEncoding.NONE,
+ DataBlockEncoding.NONE);
+ }
+ LOG.info("Using " + encoder.getEncodingInCache() + " in cache and " +
+ encoder.getEncodingOnDisk() + " on disk for the column family " +
+ Bytes.toString(family));
/* new bloom filter does not require maxKeys. */
int maxKeys = 0;
wl.writer = new StoreFile.WriterBuilder(conf, new CacheConfig(conf),
@@ -187,6 +207,7 @@ public class HFileOutputFormat extends F
.withCompression(compression)
.withBloomType(bloomType)
.withMaxKeyCount(maxKeys)
+ .withDataBlockEncoder(encoder)
.build();
this.writers.put(family, wl);
return wl;
@@ -293,7 +314,6 @@ public class HFileOutputFormat extends F
LOG.warn("Set up the HFileOutputFormat as MapperOutputFormat." +
"It is the mapper task's responsibility to make sure that each mapper emits values " +
"for one region only and in a sorted order. !");
-
Configuration conf = job.getConfiguration();
if (!KeyValue.class.equals(job.getMapOutputValueClass())) {
LOG.error("Only support the KeyValue.class as MapOutputValueClass so far!");
@@ -305,7 +325,10 @@ public class HFileOutputFormat extends F
// Set BloomFilter type based on column families and
// relevant parameters.
configureBloomFilter(table, conf);
-
+
+ // Configure the DeltaEncoding per family
+ configureDeltaEncoding(table, conf);
+
LOG.info("Configured the HFileOutputFormat as MapperOutputFormat for table: " +
table.getTableDescriptor().getNameAsString());
}
@@ -373,6 +396,9 @@ public class HFileOutputFormat extends F
// Set BloomFilter type based on column families and
// relevant parameters.
configureBloomFilter(table, conf);
+
+ // Configure the DeltaEncoding per family
+ configureDeltaEncoding(table, conf);
LOG.info("Incremental table output configured.");
}
@@ -498,4 +524,66 @@ public class HFileOutputFormat extends F
}
return bloomTypeMap;
}
+
+ protected static void configureDeltaEncoding(HTable table, Configuration conf)
+ throws IOException {
+ HTableDescriptor tableDescriptor = table.getTableDescriptor();
+ if (tableDescriptor == null){
+ return;
+ }
+
+ StringBuilder encodingTypePerCFConfigValue = new StringBuilder();
+ Collection<HColumnDescriptor> families = tableDescriptor.getFamilies();
+ int i = 0;
+ for (HColumnDescriptor familyDescriptor : families) {
+ if (i++ > 0) {
+ encodingTypePerCFConfigValue.append('&');
+ }
+
+ encodingTypePerCFConfigValue.append(
+ URLEncoder.encode(familyDescriptor.getNameAsString(), UTF8));
+ encodingTypePerCFConfigValue.append('=');
+ encodingTypePerCFConfigValue.append(
+ URLEncoder.encode(familyDescriptor.getDataBlockEncodingOnDisk().toString(),
+ UTF8));
+ encodingTypePerCFConfigValue.append(':');
+ encodingTypePerCFConfigValue.append(
+ URLEncoder.encode(familyDescriptor.getDataBlockEncoding().toString(),
+ UTF8));
+ }
+
+ conf.set(ENCODING_TYPE_PER_CF_KEY, encodingTypePerCFConfigValue.toString());
+ }
+
+ static Map<byte[], HFileDataBlockEncoder> createFamilyDeltaEncodingMap(Configuration conf) {
+ Map<byte[], HFileDataBlockEncoder> encodingTypeMap =
+ new TreeMap<byte[], HFileDataBlockEncoder >(Bytes.BYTES_COMPARATOR);
+ String encodingTypeConf = conf.get(ENCODING_TYPE_PER_CF_KEY, "");
+
+ if (encodingTypeConf.isEmpty()) {
+ return encodingTypeMap;
+ }
+
+ for (String familyConf : encodingTypeConf.split("&")) {
+ String[] familySplit = familyConf.split("=");
+ if (familySplit.length != 2) {
+ throw new AssertionError("Invalid Encoding type per family configuration");
+ }
+
+ try {
+ String[] encodingsForFamily =
+ URLDecoder.decode(familySplit[1], UTF8).split(":");
+ HFileDataBlockEncoder encoder = new HFileDataBlockEncoderImpl(
+ DataBlockEncoding.valueOf(URLDecoder.decode(encodingsForFamily[0], UTF8)),
+ DataBlockEncoding.valueOf(URLDecoder.decode(encodingsForFamily[1], UTF8)));
+ encodingTypeMap.put(
+ URLDecoder.decode(familySplit[0], UTF8).getBytes(),
+ encoder);
+ } catch (UnsupportedEncodingException e) {
+ // will not happen with UTF-8 encoding
+ throw new AssertionError(e);
+ }
+ }
+ return encodingTypeMap;
+ }
}
\ No newline at end of file
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java?rev=1432708&r1=1432707&r2=1432708&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java Sun Jan 13 19:21:08 2013
@@ -1615,7 +1615,7 @@ public class StoreFile extends SchemaCon
return reader.getTrailer().getVersion();
}
- HFile.Reader getHFileReader() {
+ public HFile.Reader getHFileReader() {
return reader;
}
Modified: hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/mapreduce/TestHFileOutputFormat.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/mapreduce/TestHFileOutputFormat.java?rev=1432708&r1=1432707&r2=1432708&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/mapreduce/TestHFileOutputFormat.java (original)
+++ hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/mapreduce/TestHFileOutputFormat.java Sun Jan 13 19:21:08 2013
@@ -56,9 +56,11 @@ import org.apache.hadoop.hbase.client.Ro
import org.apache.hadoop.hbase.client.RowMutation;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.io.hfile.Compression;
import org.apache.hadoop.hbase.io.hfile.HFile;
+import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoder;
import org.apache.hadoop.hbase.io.hfile.HFileScanner;
import org.apache.hadoop.hbase.io.hfile.Compression.Algorithm;
import org.apache.hadoop.hbase.io.hfile.HFile.Reader;
@@ -880,6 +882,111 @@ public class TestHFileOutputFormat {
}
}
+ private void setupColumnFamiliesEncodingType(HTable table,
+ Map<String, DataBlockEncoding> familyToEncoding) throws IOException {
+ HTableDescriptor mockTableDesc = new HTableDescriptor();
+ for (Entry<String, DataBlockEncoding> entry : familyToEncoding.entrySet()) {
+ mockTableDesc.addFamily(
+ new HColumnDescriptor(entry.getKey().getBytes(),
+ 1,
+ Compression.Algorithm.NONE.toString(),
+ true,
+ entry.getValue().toString(),
+ false,
+ false,
+ HColumnDescriptor.DEFAULT_BLOCKSIZE,
+ 0,
+ BloomType.NONE.toString(),
+ HColumnDescriptor.DEFAULT_REPLICATION_SCOPE,
+ HColumnDescriptor.DEFAULT_BLOOMFILTER_ERROR_RATE));
+ }
+ Mockito.doReturn(mockTableDesc).when(table).getTableDescriptor();
+ }
+
+ /**
+ * Test that {@link HFileOutputFormat} RecordWriter uses encoding settings
+ * from the column family descriptor
+ */
+ @Test
+ public void testColumnFamilyEncoding()
+ throws IOException, InterruptedException {
+ Configuration conf = new Configuration(this.util.getConfiguration());
+ RecordWriter<ImmutableBytesWritable, KeyValue> writer = null;
+ TaskAttemptContext context = null;
+ Path dir =
+ util.getTestDir("testColumnFamilyEncoding");
+
+ HTable table = Mockito.mock(HTable.class);
+
+ Map<String, DataBlockEncoding> configuredEncoding =
+ new HashMap<String, DataBlockEncoding>();
+ DataBlockEncoding[] EncodingTypeValues = DataBlockEncoding.values();
+
+ int familyIndex = 0;
+ for (byte[] family : FAMILIES) {
+ configuredEncoding.put(Bytes.toString(family),
+ EncodingTypeValues[familyIndex++ % EncodingTypeValues.length]);
+ }
+
+ setupColumnFamiliesEncodingType(table, configuredEncoding);
+
+ // set up the table to return some mock keys
+ setupMockStartKeys(table);
+
+ try {
+ // partial map red setup to get an operational writer for testing
+ Job job = new Job(conf, "testLocalMRIncrementalLoad");
+ setupRandomGeneratorMapper(job);
+ HFileOutputFormat.configureIncrementalLoad(job, table);
+ FileOutputFormat.setOutputPath(job, dir);
+ context = new TaskAttemptContext(job.getConfiguration(),
+ new TaskAttemptID());
+ HFileOutputFormat hof = new HFileOutputFormat();
+ writer = hof.getRecordWriter(context);
+
+ // write out random rows
+ writeRandomKeyValues(writer, context, ROWSPERSPLIT);
+ writer.close(context);
+
+ // Make sure that a directory was created for every CF
+ FileSystem fileSystem = dir.getFileSystem(conf);
+
+ // commit so that the filesystem has one directory per column family
+ hof.getOutputCommitter(context).commitTask(context);
+ for (byte[] family : FAMILIES) {
+ String familyStr = Bytes.toString(family);
+ boolean found = false;
+ for (FileStatus f : fileSystem.listStatus(dir)) {
+
+ if (Bytes.toString(family).equals(f.getPath().getName())) {
+ // we found a matching directory
+ found = true;
+
+ // verify that the Encoding type on this file matches the
+ // configured Encoding type.
+ Path dataFilePath = fileSystem.listStatus(f.getPath())[0].getPath();
+ StoreFile.Reader reader = new StoreFile.Reader(fileSystem,
+ dataFilePath, new CacheConfig(conf), null);
+ Map<byte[], byte[]> metadataMap = reader.loadFileInfo();
+
+ assertTrue("timeRange is not set", metadataMap.get(StoreFile.TIMERANGE_KEY) != null);
+ assertEquals("Incorrect Encoding Type used for column family "
+ + familyStr + "(reader: " + reader + ")",
+ configuredEncoding.get(familyStr),
+ reader.getHFileReader().getEncodingOnDisk());
+ break;
+ }
+ }
+
+ if (!found) {
+ fail("HFile for column family " + familyStr + " not found");
+ }
+ }
+
+ } finally {
+ dir.getFileSystem(conf).delete(dir, true);
+ }
+ }
/**
* Write random values to the writer assuming a table created using