Skip to content

Commit 4f9fbd8

Browse files
BukrosSzabolcsapurtell
BukrosSzabolcs
authored and
apurtell
committed
HBASE-26707: Reduce number of renames during bulkload (#4066) (#4122)
Signed-off-by: Wellington Ramos Chevreuil <wchevreuil@apache.org> Conflicts: hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkloadBase.java
1 parent c76c083 commit 4f9fbd8

9 files changed

Lines changed: 503 additions & 39 deletions

File tree

‎hbase-it/src/test/java/org/apache/hadoop/hbase/mapreduce/IntegrationTestFileBasedSFTBulkLoad.java‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
/**
2+
*
3+
* Licensed to the Apache Software Foundation (ASF) under one
4+
* or more contributor license agreements. See the NOTICE file
5+
* distributed with this work for additional information
6+
* regarding copyright ownership. The ASF licenses this file
7+
* to you under the Apache License, Version 2.0 (the
8+
* "License"); you may not use this file except in compliance
9+
* with the License. You may obtain a copy of the License at
10+
*
11+
* http://www.apache.org/licenses/LICENSE-2.0
12+
*
13+
* Unless required by applicable law or agreed to in writing, software
14+
* distributed under the License is distributed on an "AS IS" BASIS,
15+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16+
* See the License for the specific language governing permissions and
17+
* limitations under the License.
18+
*/
19+
package org.apache.hadoop.hbase.mapreduce;
20+
21+
import org.apache.hadoop.conf.Configuration;
22+
import org.apache.hadoop.hbase.HBaseConfiguration;
23+
import org.apache.hadoop.hbase.IntegrationTestingUtility;
24+
import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
25+
import org.apache.hadoop.hbase.testclassification.IntegrationTests;
26+
import org.apache.hadoop.util.ToolRunner;
27+
import org.junit.Test;
28+
import org.junit.experimental.categories.Category;
29+
import org.slf4j.Logger;
30+
import org.slf4j.LoggerFactory;
31+
32+
/**
33+
* Test Bulk Load and MR on a distributed cluster.
34+
* With FileBased StorefileTracker enabled.
35+
* It starts an MR job that creates linked chains
36+
*
37+
* The format of rows is like this:
38+
* Row Key -> Long
39+
*
40+
* L:<< Chain Id >> -> Row Key of the next link in the chain
41+
* S:<< Chain Id >> -> The step in the chain that his link is.
42+
* D:<< Chain Id >> -> Random Data.
43+
*
44+
* All chains start on row 0.
45+
* All rk's are > 0.
46+
*
47+
* After creating the linked lists they are walked over using a TableMapper based Mapreduce Job.
48+
*
49+
* There are a few options exposed:
50+
*
51+
* hbase.IntegrationTestBulkLoad.chainLength
52+
* The number of rows that will be part of each and every chain.
53+
*
54+
* hbase.IntegrationTestBulkLoad.numMaps
55+
* The number of mappers that will be run. Each mapper creates on linked list chain.
56+
*
57+
* hbase.IntegrationTestBulkLoad.numImportRounds
58+
* How many jobs will be run to create linked lists.
59+
*
60+
* hbase.IntegrationTestBulkLoad.tableName
61+
* The name of the table.
62+
*
63+
* hbase.IntegrationTestBulkLoad.replicaCount
64+
* How many region replicas to configure for the table under test.
65+
*/
66+
@Category(IntegrationTests.class)
67+
public class IntegrationTestFileBasedSFTBulkLoad extends IntegrationTestBulkLoad {
68+
69+
private static final Logger LOG = LoggerFactory.getLogger(IntegrationTestFileBasedSFTBulkLoad.class);
70+
71+
private static String NUM_MAPS_KEY = "hbase.IntegrationTestBulkLoad.numMaps";
72+
private static String NUM_IMPORT_ROUNDS_KEY = "hbase.IntegrationTestBulkLoad.numImportRounds";
73+
private static String NUM_REPLICA_COUNT_KEY = "hbase.IntegrationTestBulkLoad.replicaCount";
74+
private static int NUM_REPLICA_COUNT_DEFAULT = 1;
75+
76+
@Test
77+
public void testFileBasedSFTBulkLoad() throws Exception {
78+
super.testBulkLoad();
79+
}
80+
81+
@Override
82+
public void setUpCluster() throws Exception {
83+
util = getTestingUtil(getConf());
84+
util.getConfiguration().set(StoreFileTrackerFactory.TRACKER_IMPL,
85+
"org.apache.hadoop.hbase.regionserver.storefiletracker.FileBasedStoreFileTracker");
86+
util.initializeCluster(1);
87+
int replicaCount = getConf().getInt(NUM_REPLICA_COUNT_KEY, NUM_REPLICA_COUNT_DEFAULT);
88+
if (LOG.isDebugEnabled() && replicaCount != NUM_REPLICA_COUNT_DEFAULT) {
89+
LOG.debug("Region Replicas enabled: " + replicaCount);
90+
}
91+
92+
// Scale this up on a real cluster
93+
if (util.isDistributedCluster()) {
94+
util.getConfiguration().setIfUnset(NUM_MAPS_KEY,
95+
Integer.toString(util.getAdmin().getRegionServers().size() * 10));
96+
util.getConfiguration().setIfUnset(NUM_IMPORT_ROUNDS_KEY, "5");
97+
} else {
98+
util.startMiniMapReduceCluster();
99+
}
100+
}
101+
102+
public static void main(String[] args) throws Exception {
103+
Configuration conf = HBaseConfiguration.create();
104+
IntegrationTestingUtility.setUseDistributedCluster(conf);
105+
int status = ToolRunner.run(conf, new IntegrationTestFileBasedSFTBulkLoad(), args);
106+
System.exit(status);
107+
}
108+
}

‎hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java‎

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -6847,7 +6847,7 @@ public interface BulkLoadListener {
68476847
* @return final path to be used for actual loading
68486848
* @throws IOException
68496849
*/
6850-
String prepareBulkLoad(byte[] family, String srcPath, boolean copyFile)
6850+
String prepareBulkLoad(byte[] family, String srcPath, boolean copyFile, String customStaging)
68516851
throws IOException;
68526852

68536853
/**
@@ -6969,12 +6969,21 @@ public Map<byte[], List<Path>> bulkLoadHFiles(Collection<Pair<byte[], String>> f
69696969
familyWithFinalPath.put(familyName, new ArrayList<>());
69706970
}
69716971
List<Pair<Path, Path>> lst = familyWithFinalPath.get(familyName);
6972+
String finalPath = path;
69726973
try {
6973-
String finalPath = path;
6974+
boolean reqTmp = store.storeEngine.requireWritingToTmpDirFirst();
69746975
if (bulkLoadListener != null) {
6975-
finalPath = bulkLoadListener.prepareBulkLoad(familyName, path, copyFile);
6976+
finalPath = bulkLoadListener.prepareBulkLoad(familyName, path, copyFile,
6977+
reqTmp ? null : regionDir.toString());
6978+
}
6979+
Pair<Path, Path> pair = null;
6980+
if (reqTmp) {
6981+
pair = store.preBulkLoadHFile(finalPath, seqId);
6982+
}
6983+
else {
6984+
Path livePath = new Path(finalPath);
6985+
pair = new Pair<>(livePath, livePath);
69766986
}
6977-
Pair<Path, Path> pair = store.preBulkLoadHFile(finalPath, seqId);
69786987
lst.add(pair);
69796988
} catch (IOException ioe) {
69806989
// A failure here can cause an atomicity violation that we currently
@@ -6984,7 +6993,7 @@ public Map<byte[], List<Path>> bulkLoadHFiles(Collection<Pair<byte[], String>> f
69846993
" load " + Bytes.toString(p.getFirst()) + " : " + p.getSecond(), ioe);
69856994
if (bulkLoadListener != null) {
69866995
try {
6987-
bulkLoadListener.failedBulkLoad(familyName, path);
6996+
bulkLoadListener.failedBulkLoad(familyName, finalPath);
69886997
} catch (Exception ex) {
69896998
LOG.error("Error while calling failedBulkLoad for family " +
69906999
Bytes.toString(familyName) + " with path " + path, ex);

‎hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -508,6 +508,10 @@ private Path preCommitStoreFile(final String familyName, final Path buildPath,
508508
* @throws IOException
509509
*/
510510
Path commitStoreFile(final Path buildPath, Path dstPath) throws IOException {
511+
// rename is not necessary in case of direct-insert stores
512+
if(buildPath.equals(dstPath)){
513+
return dstPath;
514+
}
511515
// buildPath exists, therefore not doing an exists() check.
512516
if (!rename(buildPath, dstPath)) {
513517
throw new IOException("Failed rename of " + buildPath + " to " + dstPath);

‎hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SecureBulkLoadManager.java‎

Lines changed: 41 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import java.util.concurrent.ConcurrentHashMap;
2929
import java.util.function.Consumer;
3030

31+
import org.apache.commons.lang3.StringUtils;
3132
import org.apache.commons.lang3.mutable.MutableInt;
3233
import org.apache.hadoop.conf.Configuration;
3334
import org.apache.hadoop.fs.FileStatus;
@@ -342,27 +343,37 @@ private User getActiveUser() throws IOException {
342343
return user;
343344
}
344345

345-
private static class SecureBulkLoadListener implements BulkLoadListener {
346+
//package-private for test purpose only
347+
static class SecureBulkLoadListener implements BulkLoadListener {
346348
// Target filesystem
347349
private final FileSystem fs;
348350
private final String stagingDir;
349351
private final Configuration conf;
350352
// Source filesystem
351353
private FileSystem srcFs = null;
352354
private Map<String, FsPermission> origPermissions = null;
355+
private Map<String, String> origSources = null;
353356

354357
public SecureBulkLoadListener(FileSystem fs, String stagingDir, Configuration conf) {
355358
this.fs = fs;
356359
this.stagingDir = stagingDir;
357360
this.conf = conf;
358361
this.origPermissions = new HashMap<>();
362+
this.origSources = new HashMap<>();
359363
}
360364

361365
@Override
362-
public String prepareBulkLoad(final byte[] family, final String srcPath, boolean copyFile)
363-
throws IOException {
366+
public String prepareBulkLoad(final byte[] family, final String srcPath, boolean copyFile,
367+
String customStaging ) throws IOException {
364368
Path p = new Path(srcPath);
365-
Path stageP = new Path(stagingDir, new Path(Bytes.toString(family), p.getName()));
369+
370+
//store customStaging for failedBulkLoad
371+
String currentStaging = stagingDir;
372+
if(StringUtils.isNotEmpty(customStaging)){
373+
currentStaging = customStaging;
374+
}
375+
376+
Path stageP = new Path(currentStaging, new Path(Bytes.toString(family), p.getName()));
366377

367378
// In case of Replication for bulk load files, hfiles are already copied in staging directory
368379
if (p.equals(stageP)) {
@@ -391,11 +402,16 @@ public String prepareBulkLoad(final byte[] family, final String srcPath, boolean
391402
LOG.debug("Moving " + p + " to " + stageP);
392403
FileStatus origFileStatus = fs.getFileStatus(p);
393404
origPermissions.put(srcPath, origFileStatus.getPermission());
405+
origSources.put(stageP.toString(), srcPath);
394406
if(!fs.rename(p, stageP)) {
395407
throw new IOException("Failed to move HFile: " + p + " to " + stageP);
396408
}
397409
}
398-
fs.setPermission(stageP, PERM_ALL_ACCESS);
410+
411+
if(StringUtils.isNotEmpty(customStaging)) {
412+
fs.setPermission(stageP, PERM_ALL_ACCESS);
413+
}
414+
399415
return stageP.toString();
400416
}
401417

@@ -413,35 +429,37 @@ private void closeSrcFs() throws IOException {
413429
}
414430

415431
@Override
416-
public void failedBulkLoad(final byte[] family, final String srcPath) throws IOException {
432+
public void failedBulkLoad(final byte[] family, final String stagedPath) throws IOException {
417433
try {
418-
Path p = new Path(srcPath);
419-
if (srcFs == null) {
420-
srcFs = FileSystem.newInstance(p.toUri(), conf);
421-
}
422-
if (!FSUtils.isSameHdfs(conf, srcFs, fs)) {
423-
// files are copied so no need to move them back
434+
String src = origSources.get(stagedPath);
435+
if(StringUtils.isEmpty(src)){
436+
LOG.debug(stagedPath + " was not moved to staging. No need to move back");
424437
return;
425438
}
426-
Path stageP = new Path(stagingDir, new Path(Bytes.toString(family), p.getName()));
427439

428-
// In case of Replication for bulk load files, hfiles are not renamed by end point during
429-
// prepare stage, so no need of rename here again
430-
if (p.equals(stageP)) {
431-
LOG.debug(p.getName() + " is already available in source directory. Skipping rename.");
440+
Path stageP = new Path(stagedPath);
441+
if (!fs.exists(stageP)) {
442+
throw new IOException(
443+
"Missing HFile: " + stageP + ", can't be moved back to it's original place");
444+
}
445+
446+
//we should not move back files if the original exists
447+
Path srcPath = new Path(src);
448+
if(srcFs.exists(srcPath)) {
449+
LOG.debug(src + " is already at it's original place. No need to move.");
432450
return;
433451
}
434452

435-
LOG.debug("Moving " + stageP + " back to " + p);
436-
if (!fs.rename(stageP, p)) {
437-
throw new IOException("Failed to move HFile: " + stageP + " to " + p);
453+
LOG.debug("Moving " + stageP + " back to " + srcPath);
454+
if (!fs.rename(stageP, srcPath)) {
455+
throw new IOException("Failed to move HFile: " + stageP + " to " + srcPath);
438456
}
439457

440458
// restore original permission
441-
if (origPermissions.containsKey(srcPath)) {
442-
fs.setPermission(p, origPermissions.get(srcPath));
459+
if (origPermissions.containsKey(stagedPath)) {
460+
fs.setPermission(srcPath, origPermissions.get(src));
443461
} else {
444-
LOG.warn("Can't find previous permission for path=" + srcPath);
462+
LOG.warn("Can't find previous permission for path=" + stagedPath);
445463
}
446464
} finally {
447465
closeSrcFs();

‎hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoad.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,10 @@ public class TestBulkLoad extends TestBulkloadBase {
5454
public static final HBaseClassTestRule CLASS_RULE =
5555
HBaseClassTestRule.forClass(TestBulkLoad.class);
5656

57+
public TestBulkLoad(boolean useFileBasedSFT) {
58+
super(useFileBasedSFT);
59+
}
60+
5761
@Test
5862
public void verifyBulkLoadEvent() throws IOException {
5963
TableName tableName = TableName.valueOf("test", "test");

‎hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkloadBase.java‎

Lines changed: 33 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,9 @@
2727
import java.io.IOException;
2828
import java.util.ArrayList;
2929
import java.util.Arrays;
30+
import java.util.Collection;
3031
import java.util.List;
32+
import java.util.UUID;
3133
import org.apache.hadoop.conf.Configuration;
3234
import org.apache.hadoop.fs.FSDataOutputStream;
3335
import org.apache.hadoop.fs.Path;
@@ -44,6 +46,7 @@
4446
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
4547
import org.apache.hadoop.hbase.io.hfile.HFile;
4648
import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
49+
import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
4750
import org.apache.hadoop.hbase.util.Bytes;
4851
import org.apache.hadoop.hbase.util.Pair;
4952
import org.apache.hadoop.hbase.wal.WAL;
@@ -56,10 +59,12 @@
5659
import org.junit.Rule;
5760
import org.junit.rules.TemporaryFolder;
5861
import org.junit.rules.TestName;
59-
6062
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
6163
import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos;
64+
import org.junit.runner.RunWith;
65+
import org.junit.runners.Parameterized;
6266

67+
@RunWith(Parameterized.class)
6368
public class TestBulkloadBase {
6469
@ClassRule
6570
public static TemporaryFolder testFolder = new TemporaryFolder();
@@ -71,12 +76,31 @@ public class TestBulkloadBase {
7176
protected final byte[] family2 = Bytes.toBytes("family2");
7277
protected final byte[] family3 = Bytes.toBytes("family3");
7378

79+
protected Boolean useFileBasedSFT;
80+
7481
@Rule
7582
public TestName name = new TestName();
7683

84+
public TestBulkloadBase(boolean useFileBasedSFT) {
85+
this.useFileBasedSFT = useFileBasedSFT;
86+
}
87+
88+
@Parameterized.Parameters
89+
public static Collection<Boolean> data() {
90+
Boolean[] data = {false, true};
91+
return Arrays.asList(data);
92+
}
93+
7794
@Before
7895
public void before() throws IOException {
7996
Bytes.random(randomBytes);
97+
if(useFileBasedSFT) {
98+
conf.set(StoreFileTrackerFactory.TRACKER_IMPL,
99+
"org.apache.hadoop.hbase.regionserver.storefiletracker.FileBasedStoreFileTracker");
100+
}
101+
else {
102+
conf.unset(StoreFileTrackerFactory.TRACKER_IMPL);
103+
}
80104
}
81105

82106
protected Pair<byte[], String> withMissingHFileForFamily(byte[] family) {
@@ -111,7 +135,7 @@ protected HRegion testRegionWithFamiliesAndSpecifiedTableName(TableName tableNam
111135
}
112136

113137
protected HRegion testRegionWithFamilies(byte[]... families) throws IOException {
114-
TableName tableName = TableName.valueOf(name.getMethodName());
138+
TableName tableName = TableName.valueOf(name.getMethodName().substring(0, name.getMethodName().indexOf("[")));
115139
return testRegionWithFamiliesAndSpecifiedTableName(tableName, families);
116140
}
117141

@@ -130,7 +154,7 @@ protected List<Pair<byte[], String>> withFamilyPathsFor(byte[]... families) thro
130154
private String createHFileForFamilies(byte[] family) throws IOException {
131155
HFile.WriterFactory hFileFactory = HFile.getWriterFactoryNoCache(conf);
132156
// TODO We need a way to do this without creating files
133-
File hFileLocation = testFolder.newFile();
157+
File hFileLocation = testFolder.newFile(generateUniqueName(null));
134158
FSDataOutputStream out = new FSDataOutputStream(new FileOutputStream(hFileLocation), null);
135159
try {
136160
hFileFactory.withOutputStream(out);
@@ -149,6 +173,12 @@ private String createHFileForFamilies(byte[] family) throws IOException {
149173
return hFileLocation.getAbsoluteFile().getAbsolutePath();
150174
}
151175

176+
private static String generateUniqueName(final String suffix) {
177+
String name = UUID.randomUUID().toString().replaceAll("-", "");
178+
if (suffix != null) name += suffix;
179+
return name;
180+
}
181+
152182
protected static Matcher<WALEdit> bulkLogWalEditType(byte[] typeBytes) {
153183
return new WalMatcher(typeBytes);
154184
}

0 commit comments

Comments
 (0)