@Test public void test_LATEST_TIMESTAMP_isReplaced() throws IOException, InterruptedException { Configuration conf = new Configuration(this.util.getConfiguration()); RecordWriter<ImmutableBytesWritable, KeyValue> writer = null; TaskAttemptContext context = null; Path dir = HBaseTestingUtility.getTestDir("test_LATEST_TIMESTAMP_isReplaced"); try { Job job = new Job(conf); FileOutputFormat.setOutputPath(job, dir); context = new TaskAttemptContext(job.getConfiguration(), new TaskAttemptID()); HFileOutputFormat hof = new HFileOutputFormat(); writer = hof.getRecordWriter(context); final byte [] b = Bytes.toBytes("b"); KeyValue kv = new KeyValue(b, b, b); KeyValue original = kv.clone(); writer.write(new ImmutableBytesWritable(), kv); assertFalse(original.equals(kv)); assertTrue(Bytes.equals(original.getRow(), kv.getRow())); assertTrue(original.matchingColumn(kv.getFamily(), kv.getQualifier())); assertNotSame(original.getTimestamp(), kv.getTimestamp()); assertNotSame(HConstants.LATEST_TIMESTAMP, kv.getTimestamp()); kv = new KeyValue(b, b, b, kv.getTimestamp() - 1, b); original = kv.clone(); writer.write(new ImmutableBytesWritable(), kv); assertTrue(original.equals(kv)); } finally { if (writer != null && context != null) writer.close(context); dir.getFileSystem(conf).delete(dir, true); } }
public RecordWriter<ImmutableBytesWritable, KeyValue> getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException { final Path outputPath = FileOutputFormat.getOutputPath(context); final Path outputdir = new FileOutputCommitter(outputPath, context).getWorkPath(); Configuration conf = context.getConfiguration(); final FileSystem fs = outputdir.getFileSystem(conf); final long maxsize = conf.getLong("hbase.hregion.max.filesize", 268435456); final int blocksize = conf.getInt("hbase.mapreduce.hfileoutputformat.blocksize", 65536); final String compression = conf.get("hfile.compression", Compression.Algorithm.NONE.getName()); return new RecordWriter<ImmutableBytesWritable, KeyValue>() { private final Map<byte [], WriterLength> writers = new TreeMap<byte [], WriterLength>(Bytes.BYTES_COMPARATOR); private byte [] previousRow = HConstants.EMPTY_BYTE_ARRAY; private final byte [] now = Bytes.toBytes(System.currentTimeMillis()); public void write(ImmutableBytesWritable row, KeyValue kv) throws IOException { long length = kv.getLength(); byte [] family = kv.getFamily(); WriterLength wl = this.writers.get(family); if (wl == null || ((length + wl.written) >= maxsize) && Bytes.compareTo(this.previousRow, 0, this.previousRow.length, kv.getBuffer(), kv.getRowOffset(), kv.getRowLength()) != 0) { Path basedir = new Path(outputdir, Bytes.toString(family)); if (wl == null) { wl = new WriterLength(); this.writers.put(family, wl); if (this.writers.size() > 1) throw new IOException("One family only"); if (!fs.exists(basedir)) fs.mkdirs(basedir); } wl.writer = getNewWriter(wl.writer, basedir); LOG.info("Writer=" + wl.writer.getPath() + ((wl.written == 0)? "": ", wrote=" + wl.written)); wl.written = 0; } kv.updateLatestStamp(this.now); wl.writer.append(kv); wl.written += length; this.previousRow = kv.getRow(); } private HFile.Writer getNewWriter(final HFile.Writer writer, final Path familydir) throws IOException { close(writer); return new HFile.Writer(fs, StoreFile.getUniqueFile(fs, familydir), blocksize, compression, KeyValue.KEY_COMPARATOR); } private void close(final HFile.Writer w) throws IOException { if (w != null) { w.appendFileInfo(StoreFile.BULKLOAD_TIME_KEY, Bytes.toBytes(System.currentTimeMillis())); w.appendFileInfo(StoreFile.BULKLOAD_TASK_KEY, Bytes.toBytes(context.getTaskAttemptID().toString())); w.appendFileInfo(StoreFile.MAJOR_COMPACTION_KEY, Bytes.toBytes(true)); w.close(); } } public void close(TaskAttemptContext c) throws IOException, InterruptedException { for (Map.Entry<byte [], WriterLength> e: this.writers.entrySet()) { close(e.getValue().writer); } } }; }
HFileOutputFormat extends FileOutputFormat<ImmutableBytesWritable, KeyValue> { public RecordWriter<ImmutableBytesWritable, KeyValue> getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException { final Path outputPath = FileOutputFormat.getOutputPath(context); final Path outputdir = new FileOutputCommitter(outputPath, context).getWorkPath(); Configuration conf = context.getConfiguration(); final FileSystem fs = outputdir.getFileSystem(conf); final long maxsize = conf.getLong("hbase.hregion.max.filesize", 268435456); final int blocksize = conf.getInt("hbase.mapreduce.hfileoutputformat.blocksize", 65536); final String compression = conf.get("hfile.compression", Compression.Algorithm.NONE.getName()); return new RecordWriter<ImmutableBytesWritable, KeyValue>() { private final Map<byte [], WriterLength> writers = new TreeMap<byte [], WriterLength>(Bytes.BYTES_COMPARATOR); private byte [] previousRow = HConstants.EMPTY_BYTE_ARRAY; private final byte [] now = Bytes.toBytes(System.currentTimeMillis()); public void write(ImmutableBytesWritable row, KeyValue kv) throws IOException { long length = kv.getLength(); byte [] family = kv.getFamily(); WriterLength wl = this.writers.get(family); if (wl == null || ((length + wl.written) >= maxsize) && Bytes.compareTo(this.previousRow, 0, this.previousRow.length, kv.getBuffer(), kv.getRowOffset(), kv.getRowLength()) != 0) { Path basedir = new Path(outputdir, Bytes.toString(family)); if (wl == null) { wl = new WriterLength(); this.writers.put(family, wl); if (this.writers.size() > 1) throw new IOException("One family only"); if (!fs.exists(basedir)) fs.mkdirs(basedir); } wl.writer = getNewWriter(wl.writer, basedir); LOG.info("Writer=" + wl.writer.getPath() + ((wl.written == 0)? "": ", wrote=" + wl.written)); wl.written = 0; } kv.updateLatestStamp(this.now); wl.writer.append(kv); wl.written += length; this.previousRow = kv.getRow(); } private HFile.Writer getNewWriter(final HFile.Writer writer, final Path familydir) throws IOException { close(writer); return new HFile.Writer(fs, StoreFile.getUniqueFile(fs, familydir), blocksize, compression, KeyValue.KEY_COMPARATOR); } private void close(final HFile.Writer w) throws IOException { if (w != null) { w.appendFileInfo(StoreFile.BULKLOAD_TIME_KEY, Bytes.toBytes(System.currentTimeMillis())); w.appendFileInfo(StoreFile.BULKLOAD_TASK_KEY, Bytes.toBytes(context.getTaskAttemptID().toString())); w.appendFileInfo(StoreFile.MAJOR_COMPACTION_KEY, Bytes.toBytes(true)); w.close(); } } public void close(TaskAttemptContext c) throws IOException, InterruptedException { for (Map.Entry<byte [], WriterLength> e: this.writers.entrySet()) { close(e.getValue().writer); } } }; } }
HFileOutputFormat extends FileOutputFormat<ImmutableBytesWritable, KeyValue> { public RecordWriter<ImmutableBytesWritable, KeyValue> getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException { final Path outputPath = FileOutputFormat.getOutputPath(context); final Path outputdir = new FileOutputCommitter(outputPath, context).getWorkPath(); Configuration conf = context.getConfiguration(); final FileSystem fs = outputdir.getFileSystem(conf); final long maxsize = conf.getLong("hbase.hregion.max.filesize", 268435456); final int blocksize = conf.getInt("hbase.mapreduce.hfileoutputformat.blocksize", 65536); final String compression = conf.get("hfile.compression", Compression.Algorithm.NONE.getName()); return new RecordWriter<ImmutableBytesWritable, KeyValue>() { private final Map<byte [], WriterLength> writers = new TreeMap<byte [], WriterLength>(Bytes.BYTES_COMPARATOR); private byte [] previousRow = HConstants.EMPTY_BYTE_ARRAY; private final byte [] now = Bytes.toBytes(System.currentTimeMillis()); public void write(ImmutableBytesWritable row, KeyValue kv) throws IOException { long length = kv.getLength(); byte [] family = kv.getFamily(); WriterLength wl = this.writers.get(family); if (wl == null || ((length + wl.written) >= maxsize) && Bytes.compareTo(this.previousRow, 0, this.previousRow.length, kv.getBuffer(), kv.getRowOffset(), kv.getRowLength()) != 0) { Path basedir = new Path(outputdir, Bytes.toString(family)); if (wl == null) { wl = new WriterLength(); this.writers.put(family, wl); if (this.writers.size() > 1) throw new IOException("One family only"); if (!fs.exists(basedir)) fs.mkdirs(basedir); } wl.writer = getNewWriter(wl.writer, basedir); LOG.info("Writer=" + wl.writer.getPath() + ((wl.written == 0)? "": ", wrote=" + wl.written)); wl.written = 0; } kv.updateLatestStamp(this.now); wl.writer.append(kv); wl.written += length; this.previousRow = kv.getRow(); } private HFile.Writer getNewWriter(final HFile.Writer writer, final Path familydir) throws IOException { close(writer); return new HFile.Writer(fs, StoreFile.getUniqueFile(fs, familydir), blocksize, compression, KeyValue.KEY_COMPARATOR); } private void close(final HFile.Writer w) throws IOException { if (w != null) { w.appendFileInfo(StoreFile.BULKLOAD_TIME_KEY, Bytes.toBytes(System.currentTimeMillis())); w.appendFileInfo(StoreFile.BULKLOAD_TASK_KEY, Bytes.toBytes(context.getTaskAttemptID().toString())); w.appendFileInfo(StoreFile.MAJOR_COMPACTION_KEY, Bytes.toBytes(true)); w.close(); } } public void close(TaskAttemptContext c) throws IOException, InterruptedException { for (Map.Entry<byte [], WriterLength> e: this.writers.entrySet()) { close(e.getValue().writer); } } }; }  }
HFileOutputFormat extends FileOutputFormat<ImmutableBytesWritable, KeyValue> { public RecordWriter<ImmutableBytesWritable, KeyValue> getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException { final Path outputPath = FileOutputFormat.getOutputPath(context); final Path outputdir = new FileOutputCommitter(outputPath, context).getWorkPath(); Configuration conf = context.getConfiguration(); final FileSystem fs = outputdir.getFileSystem(conf); final long maxsize = conf.getLong("hbase.hregion.max.filesize", 268435456); final int blocksize = conf.getInt("hbase.mapreduce.hfileoutputformat.blocksize", 65536); final String compression = conf.get("hfile.compression", Compression.Algorithm.NONE.getName()); return new RecordWriter<ImmutableBytesWritable, KeyValue>() { private final Map<byte [], WriterLength> writers = new TreeMap<byte [], WriterLength>(Bytes.BYTES_COMPARATOR); private byte [] previousRow = HConstants.EMPTY_BYTE_ARRAY; private final byte [] now = Bytes.toBytes(System.currentTimeMillis()); public void write(ImmutableBytesWritable row, KeyValue kv) throws IOException { long length = kv.getLength(); byte [] family = kv.getFamily(); WriterLength wl = this.writers.get(family); if (wl == null || ((length + wl.written) >= maxsize) && Bytes.compareTo(this.previousRow, 0, this.previousRow.length, kv.getBuffer(), kv.getRowOffset(), kv.getRowLength()) != 0) { Path basedir = new Path(outputdir, Bytes.toString(family)); if (wl == null) { wl = new WriterLength(); this.writers.put(family, wl); if (this.writers.size() > 1) throw new IOException("One family only"); if (!fs.exists(basedir)) fs.mkdirs(basedir); } wl.writer = getNewWriter(wl.writer, basedir); LOG.info("Writer=" + wl.writer.getPath() + ((wl.written == 0)? "": ", wrote=" + wl.written)); wl.written = 0; } kv.updateLatestStamp(this.now); wl.writer.append(kv); wl.written += length; this.previousRow = kv.getRow(); } private HFile.Writer getNewWriter(final HFile.Writer writer, final Path familydir) throws IOException { close(writer); return new HFile.Writer(fs, StoreFile.getUniqueFile(fs, familydir), blocksize, compression, KeyValue.KEY_COMPARATOR); } private void close(final HFile.Writer w) throws IOException { if (w != null) { w.appendFileInfo(StoreFile.BULKLOAD_TIME_KEY, Bytes.toBytes(System.currentTimeMillis())); w.appendFileInfo(StoreFile.BULKLOAD_TASK_KEY, Bytes.toBytes(context.getTaskAttemptID().toString())); w.appendFileInfo(StoreFile.MAJOR_COMPACTION_KEY, Bytes.toBytes(true)); w.close(); } } public void close(TaskAttemptContext c) throws IOException, InterruptedException { for (Map.Entry<byte [], WriterLength> e: this.writers.entrySet()) { close(e.getValue().writer); } } }; }  RecordWriter<ImmutableBytesWritable, KeyValue> getRecordWriter(final TaskAttemptContext context); static void configureIncrementalLoad(Job job, HTable table); }
HFileOutputFormat extends FileOutputFormat<ImmutableBytesWritable, KeyValue> { public RecordWriter<ImmutableBytesWritable, KeyValue> getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException { final Path outputPath = FileOutputFormat.getOutputPath(context); final Path outputdir = new FileOutputCommitter(outputPath, context).getWorkPath(); Configuration conf = context.getConfiguration(); final FileSystem fs = outputdir.getFileSystem(conf); final long maxsize = conf.getLong("hbase.hregion.max.filesize", 268435456); final int blocksize = conf.getInt("hbase.mapreduce.hfileoutputformat.blocksize", 65536); final String compression = conf.get("hfile.compression", Compression.Algorithm.NONE.getName()); return new RecordWriter<ImmutableBytesWritable, KeyValue>() { private final Map<byte [], WriterLength> writers = new TreeMap<byte [], WriterLength>(Bytes.BYTES_COMPARATOR); private byte [] previousRow = HConstants.EMPTY_BYTE_ARRAY; private final byte [] now = Bytes.toBytes(System.currentTimeMillis()); public void write(ImmutableBytesWritable row, KeyValue kv) throws IOException { long length = kv.getLength(); byte [] family = kv.getFamily(); WriterLength wl = this.writers.get(family); if (wl == null || ((length + wl.written) >= maxsize) && Bytes.compareTo(this.previousRow, 0, this.previousRow.length, kv.getBuffer(), kv.getRowOffset(), kv.getRowLength()) != 0) { Path basedir = new Path(outputdir, Bytes.toString(family)); if (wl == null) { wl = new WriterLength(); this.writers.put(family, wl); if (this.writers.size() > 1) throw new IOException("One family only"); if (!fs.exists(basedir)) fs.mkdirs(basedir); } wl.writer = getNewWriter(wl.writer, basedir); LOG.info("Writer=" + wl.writer.getPath() + ((wl.written == 0)? "": ", wrote=" + wl.written)); wl.written = 0; } kv.updateLatestStamp(this.now); wl.writer.append(kv); wl.written += length; this.previousRow = kv.getRow(); } private HFile.Writer getNewWriter(final HFile.Writer writer, final Path familydir) throws IOException { close(writer); return new HFile.Writer(fs, StoreFile.getUniqueFile(fs, familydir), blocksize, compression, KeyValue.KEY_COMPARATOR); } private void close(final HFile.Writer w) throws IOException { if (w != null) { w.appendFileInfo(StoreFile.BULKLOAD_TIME_KEY, Bytes.toBytes(System.currentTimeMillis())); w.appendFileInfo(StoreFile.BULKLOAD_TASK_KEY, Bytes.toBytes(context.getTaskAttemptID().toString())); w.appendFileInfo(StoreFile.MAJOR_COMPACTION_KEY, Bytes.toBytes(true)); w.close(); } } public void close(TaskAttemptContext c) throws IOException, InterruptedException { for (Map.Entry<byte [], WriterLength> e: this.writers.entrySet()) { close(e.getValue().writer); } } }; }  RecordWriter<ImmutableBytesWritable, KeyValue> getRecordWriter(final TaskAttemptContext context); static void configureIncrementalLoad(Job job, HTable table);  }