You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by sh...@apache.org on 2016/04/28 12:45:44 UTC
[3/5] kylin git commit: KYLIN-1323 engine v2: split region to
multiple hfiles
KYLIN-1323 engine v2: split region to multiple hfiles
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/8e7a76b9
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/8e7a76b9
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/8e7a76b9
Branch: refs/heads/master
Commit: 8e7a76b9fc3b90ef60730dfab08d3e45b8a42675
Parents: 866fdb4
Author: shaofengshi <sh...@apache.org>
Authored: Fri Apr 1 17:32:44 2016 +0800
Committer: shaofengshi <sh...@apache.org>
Committed: Thu Apr 28 18:36:05 2016 +0800
----------------------------------------------------------------------
build/conf/kylin.properties | 15 +--
.../apache/kylin/common/KylinConfigBase.java | 20 ++--
.../org/apache/kylin/cube/util/CubingUtils.java | 1 -
.../apache/kylin/engine/spark/SparkCubing.java | 2 +-
.../test_case_data/sandbox/kylin.properties | 8 +-
.../kylin/provision/BuildCubeWithEngine.java | 4 +-
.../apache/kylin/source/hive/HiveMRInput.java | 6 --
.../storage/hbase/steps/CreateHTableJob.java | 103 ++++++++++++++++---
.../kylin/storage/hbase/steps/CubeHFileJob.java | 2 +
.../hbase/steps/RangeKeyDistributionJob.java | 11 +-
.../steps/RangeKeyDistributionReducer.java | 26 ++---
.../kylin/storage/hbase/steps/RegionSize.java | 18 +---
.../storage/hbase/steps/CreateHTableTest.java | 3 +-
13 files changed, 140 insertions(+), 79 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/build/conf/kylin.properties
----------------------------------------------------------------------
diff --git a/build/conf/kylin.properties b/build/conf/kylin.properties
index 5827fd8..80ae004 100644
--- a/build/conf/kylin.properties
+++ b/build/conf/kylin.properties
@@ -77,14 +77,11 @@ kylin.hbase.default.compression.codec=snappy
kylin.job.cubing.inmem.sampling.percent=100
# The cut size for hbase region, in GB.
-# E.g, for cube whose capacity be marked as "SMALL", split region per 5GB by default
-kylin.hbase.region.cut.small=5
-kylin.hbase.region.cut.medium=10
-kylin.hbase.region.cut.large=50
+kylin.hbase.region.cut=5
# The hfile size of GB, smaller hfile leading to the converting hfile MR has more reducers and be faster
# set 0 to disable this optimization
-kylin.hbase.hfile.size.gb=5
+kylin.hbase.hfile.size.gb=2
# Enable/disable ACL check for cube query
kylin.query.security.enabled=true
@@ -165,4 +162,10 @@ deploy.env=QA
###########################deprecated configs#######################
kylin.sandbox=true
-kylin.web.hive.limit=20
\ No newline at end of file
+kylin.web.hive.limit=20
+# The cut size for hbase region,
+#in GB.
+# E.g, for cube whose capacity be marked as "SMALL", split region per 5GB by default
+kylin.hbase.region.cut.small=5
+kylin.hbase.region.cut.medium=10
+kylin.hbase.region.cut.large=50
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
index 0e162bf..ae6382b 100644
--- a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
@@ -412,8 +412,8 @@ abstract public class KylinConfigBase implements Serializable {
setProperty("kylin.hbase.hfile.size.gb", String.valueOf(size));
}
- public int getHBaseHFileSizeGB() {
- return Integer.parseInt(getOptional("kylin.hbase.hfile.size.gb", "5"));
+ public float getHBaseHFileSizeGB() {
+ return Float.parseFloat(getOptional("kylin.hbase.hfile.size.gb", "5.0"));
}
public int getScanThreshold() {
@@ -508,16 +508,12 @@ abstract public class KylinConfigBase implements Serializable {
return Boolean.parseBoolean(this.getOptional("kylin.dict.growing.enabled", "false"));
}
- public float getKylinHBaseRegionCutSmall() {
- return Float.valueOf(getOptional("kylin.hbase.region.cut.small", "10"));
- }
-
- public float getKylinHBaseRegionCutMedium() {
- return Float.valueOf(getOptional("kylin.hbase.region.cut.medium", "20"));
- }
-
- public float getKylinHBaseRegionCutLarge() {
- return Float.valueOf(getOptional("kylin.hbase.region.cut.large", "100"));
+ /**
+ * HBase region cut size, in GB
+ * @return
+ */
+ public float getKylinHBaseRegionCut() {
+ return Float.valueOf(getOptional("kylin.hbase.region.cut", "1.0"));
}
public int getHBaseScanMaxResultSize() {
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/core-cube/src/main/java/org/apache/kylin/cube/util/CubingUtils.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/util/CubingUtils.java b/core-cube/src/main/java/org/apache/kylin/cube/util/CubingUtils.java
index ac81318..b33062d 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/util/CubingUtils.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/util/CubingUtils.java
@@ -41,7 +41,6 @@ import java.util.Map;
import javax.annotation.Nullable;
-import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeSegment;
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
----------------------------------------------------------------------
diff --git a/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java b/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
index 70c1032..2b4e497 100644
--- a/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
+++ b/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
@@ -477,7 +477,7 @@ public class SparkCubing extends AbstractApplication {
final Map<Long, Long> rowCountMap = CubeStatsReader.getCuboidRowCountMapFromSampling(samplingResult, 100);
final Map<Long, Double> cubeSizeMap = CubeStatsReader.getCuboidSizeMapFromRowCount(cubeSegment, rowCountMap);
System.out.println("cube size estimation:" + cubeSizeMap);
- final byte[][] splitKeys = CreateHTableJob.getSplitsFromCuboidStatistics(cubeSizeMap, kylinConfig, cubeSegment);
+ final byte[][] splitKeys = CreateHTableJob.getRegionSplitsFromCuboidStatistics(cubeSizeMap, kylinConfig, cubeSegment, null); //FIXME: passing non-null value for 'hfileSplitsOutputFolder'
CubeHTableUtil.createHTable(cubeSegment, splitKeys);
System.out.println(cubeSegment.getStorageLocationIdentifier() + " table created");
return splitKeys;
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/examples/test_case_data/sandbox/kylin.properties
----------------------------------------------------------------------
diff --git a/examples/test_case_data/sandbox/kylin.properties b/examples/test_case_data/sandbox/kylin.properties
index cf1f881..a004266 100644
--- a/examples/test_case_data/sandbox/kylin.properties
+++ b/examples/test_case_data/sandbox/kylin.properties
@@ -69,9 +69,11 @@ kylin.job.cubing.inmem.sampling.percent=100
# The cut size for hbase region, in GB.
# E.g, for cube whose capacity be marked as "SMALL", split region per 10GB by default
-kylin.hbase.region.cut.small=10
-kylin.hbase.region.cut.medium=20
-kylin.hbase.region.cut.large=100
+kylin.hbase.region.cut=5
+# The hfile size of GB, smaller hfile leading to the converting hfile MR has more reducers and be faster
+# set 0 to disable this optimization
+kylin.hbase.hfile.size.gb=2
+
# Time interval to check hadoop job status
kylin.job.yarn.app.rest.check.interval.seconds=10
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java
index 71bb34f..0a5ba5a 100644
--- a/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java
+++ b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java
@@ -357,9 +357,7 @@ public class BuildCubeWithEngine {
DefaultChainedExecutable job = EngineFactory.createBatchCubingJob(segment, "TEST");
jobService.addJob(job);
waitForJob(job.getId());
- if (segment.getCubeDesc().getEngineType() == IEngineAware.ID_MR_V1) {
- checkHFilesInHBase(segment);
- }
+ checkHFilesInHBase(segment);
return job.getId();
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
----------------------------------------------------------------------
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
index 5242d76..7088168 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
@@ -23,19 +23,13 @@ import java.io.IOException;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hive.ql.metadata.Hive;
-import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hive.hcatalog.data.HCatRecord;
import org.apache.hive.hcatalog.mapreduce.HCatInputFormat;
import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.model.CubeJoinedFlatTableDesc;
import org.apache.kylin.engine.mr.HadoopUtil;
import org.apache.kylin.engine.mr.IMRInput;
import org.apache.kylin.engine.mr.JobBuilderSupport;
-import org.apache.kylin.invertedindex.IISegment;
-import org.apache.kylin.invertedindex.model.IIJoinedFlatTableDesc;
import org.apache.kylin.job.JoinedFlatTable;
import org.apache.kylin.job.common.ShellExecutable;
import org.apache.kylin.job.constant.ExecutableConstants;
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CreateHTableJob.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CreateHTableJob.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CreateHTableJob.java
index ef5cc38..67a9c4f 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CreateHTableJob.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CreateHTableJob.java
@@ -29,10 +29,9 @@ import org.apache.commons.cli.Options;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.io.SequenceFile;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.ToolRunner;
@@ -44,10 +43,13 @@ import org.apache.kylin.common.util.ShardingHash;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
+import org.apache.kylin.engine.mr.HadoopUtil;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.CubeStatsReader;
import org.apache.kylin.engine.mr.common.CuboidShardUtil;
+import org.apache.kylin.metadata.model.DataModelDesc;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
import org.apache.kylin.storage.hbase.HBaseConnection;
import org.slf4j.Logger;
@@ -66,6 +68,7 @@ public class CreateHTableJob extends AbstractHadoopJob {
CubeDesc cubeDesc = null;
String segmentName = null;
KylinConfig kylinConfig;
+ Path partitionFilePath;
@Override
public int run(String[] args) throws Exception {
@@ -77,7 +80,7 @@ public class CreateHTableJob extends AbstractHadoopJob {
options.addOption(OPTION_STATISTICS_ENABLED);
parseOptions(options, args);
- Path partitionFilePath = new Path(getOptionValue(OPTION_PARTITION_FILE_PATH));
+ partitionFilePath = new Path(getOptionValue(OPTION_PARTITION_FILE_PATH));
boolean statsEnabled = Boolean.parseBoolean(getOptionValue(OPTION_STATISTICS_ENABLED));
String cubeName = getOptionValue(OPTION_CUBE_NAME).toUpperCase();
@@ -94,9 +97,9 @@ public class CreateHTableJob extends AbstractHadoopJob {
byte[][] splitKeys;
if (statsEnabled) {
final Map<Long, Double> cuboidSizeMap = new CubeStatsReader(cubeSegment, kylinConfig).getCuboidSizeMap();
- splitKeys = getSplitsFromCuboidStatistics(cuboidSizeMap, kylinConfig, cubeSegment);
+ splitKeys = getRegionSplitsFromCuboidStatistics(cuboidSizeMap, kylinConfig, cubeSegment, partitionFilePath.getParent());
} else {
- splitKeys = getSplits(conf, partitionFilePath);
+ splitKeys = getRegionSplits(conf, partitionFilePath);
}
CubeHTableUtil.createHTable(cubeSegment, splitKeys);
@@ -110,7 +113,7 @@ public class CreateHTableJob extends AbstractHadoopJob {
}
@SuppressWarnings("deprecation")
- public byte[][] getSplits(Configuration conf, Path path) throws Exception {
+ public byte[][] getRegionSplits(Configuration conf, Path path) throws Exception {
FileSystem fs = path.getFileSystem(conf);
if (fs.exists(path) == false) {
System.err.println("Path " + path + " not found, no region split, HTable will be one region");
@@ -147,19 +150,19 @@ public class CreateHTableJob extends AbstractHadoopJob {
private static byte[][] getSplitsByRegionCount(int regionCount) {
byte[][] result = new byte[regionCount - 1][];
for (int i = 1; i < regionCount; ++i) {
- byte[] split = new byte[Bytes.SIZEOF_SHORT];
- BytesUtil.writeUnsigned(i, split, 0, Bytes.SIZEOF_SHORT);
+ byte[] split = new byte[RowConstants.ROWKEY_SHARDID_LEN];
+ BytesUtil.writeUnsigned(i, split, 0, RowConstants.ROWKEY_SHARDID_LEN);
result[i - 1] = split;
}
return result;
}
- public static byte[][] getSplitsFromCuboidStatistics(final Map<Long, Double> cubeSizeMap, KylinConfig kylinConfig, CubeSegment cubeSegment) throws IOException {
+ public static byte[][] getRegionSplitsFromCuboidStatistics(final Map<Long, Double> cubeSizeMap, final KylinConfig kylinConfig, final CubeSegment cubeSegment, final Path hfileSplitsOutputFolder) throws IOException {
final CubeDesc cubeDesc = cubeSegment.getCubeDesc();
float cut = RegionSize.getReionSize(kylinConfig, cubeDesc);
- logger.info("chosen cut for HTable is " + cut + "GB");
+ logger.info("Cut for HBase region is " + cut + "GB");
double totalSizeInM = 0;
for (Double cuboidSize : cubeSizeMap.values()) {
@@ -199,6 +202,12 @@ public class CreateHTableJob extends AbstractHadoopJob {
if (cubeSegment.isEnableSharding()) {
//each cuboid will be split into different number of shards
HashMap<Long, Short> cuboidShards = Maps.newHashMap();
+
+ //each shard/region may be split into multiple hfiles; array index: region ID, Map: key: cuboidID, value cuboid size in the region
+ List<HashMap<Long, Double>> innerRegionSplits = Lists.newArrayList();
+ for (int i = 0; i < nRegion; i++) {
+ innerRegionSplits.add(new HashMap<Long, Double>());
+ }
double[] regionSizes = new double[nRegion];
for (long cuboidId : allCuboids) {
double estimatedSize = cubeSizeMap.get(cuboidId);
@@ -220,6 +229,7 @@ public class CreateHTableJob extends AbstractHadoopJob {
for (short i = startShard; i < startShard + shardNum; ++i) {
short j = (short) (i % nRegion);
regionSizes[j] = regionSizes[j] + estimatedSize / shardNum;
+ innerRegionSplits.get(j).put(cuboidId, estimatedSize / shardNum);
}
}
@@ -228,7 +238,7 @@ public class CreateHTableJob extends AbstractHadoopJob {
}
CuboidShardUtil.saveCuboidShards(cubeSegment, cuboidShards, nRegion);
-
+ saveHFileSplits(innerRegionSplits, mbPerRegion, hfileSplitsOutputFolder, kylinConfig);
return getSplitsByRegionCount(nRegion);
} else {
@@ -260,6 +270,73 @@ public class CreateHTableJob extends AbstractHadoopJob {
}
}
+ protected static void saveHFileSplits(final List<HashMap<Long, Double>> innerRegionSplits, int mbPerRegion, final Path outputFolder, final KylinConfig kylinConfig) throws IOException {
+
+ if (outputFolder == null) {
+ logger.warn("outputFolder for hfile split file is null, skip inner region split");
+ return;
+ }
+
+ FileSystem fs = FileSystem.get(HadoopUtil.getCurrentConfiguration());
+ if (fs.exists(outputFolder) == false) {
+ fs.mkdirs(outputFolder);
+ }
+
+ float hfileSizeMB = kylinConfig.getHBaseHFileSizeGB() * 1024;
+
+ int compactionThreshold = Integer.valueOf(HBaseConnection.getCurrentHBaseConfiguration().get("hbase.hstore.compactionThreshold", "3"));
+ if (hfileSizeMB * compactionThreshold < mbPerRegion) {
+ hfileSizeMB = mbPerRegion / compactionThreshold;
+ }
+
+ if (hfileSizeMB <= 0) {
+ hfileSizeMB = mbPerRegion;
+ }
+
+ logger.info("hfileSizeMB:" + hfileSizeMB);
+ final Path hfilePartitionFile = new Path(outputFolder, "part-r-00000_hfile");
+ short regionCount = (short) innerRegionSplits.size();
+
+ List<byte[]> splits = Lists.newArrayList();
+ for (int i = 0; i < regionCount; i++) {
+ if (i > 0) {
+ // skip 0
+ byte[] split = new byte[RowConstants.ROWKEY_SHARDID_LEN];
+ BytesUtil.writeUnsigned(i, split, 0, RowConstants.ROWKEY_SHARDID_LEN);
+ splits.add(split); // split by region;
+ }
+
+ HashMap<Long, Double> cuboidSize = innerRegionSplits.get(i);
+ List<Long> allCuboids = Lists.newArrayList();
+ allCuboids.addAll(cuboidSize.keySet());
+ Collections.sort(allCuboids);
+
+ double accumulatedSize = 0;
+ int j = 0;
+ for (Long cuboid : allCuboids) {
+ if (accumulatedSize >= hfileSizeMB) {
+ logger.info(String.format("Region %d's hfile %d size is %.2f mb", i, j, accumulatedSize));
+ byte[] split = new byte[RowConstants.ROWKEY_SHARD_AND_CUBOID_LEN];
+ BytesUtil.writeUnsigned(i, split, 0, RowConstants.ROWKEY_SHARDID_LEN);
+ System.arraycopy(Bytes.toBytes(cuboid), 0, split, RowConstants.ROWKEY_SHARDID_LEN, RowConstants.ROWKEY_CUBOIDID_LEN);
+ splits.add(split);
+ accumulatedSize = 0;
+ j++;
+ }
+ accumulatedSize += cuboidSize.get(cuboid);
+ }
+
+ }
+
+ Configuration conf = HadoopUtil.getCurrentConfiguration();
+ SequenceFile.Writer hfilePartitionWriter = SequenceFile.createWriter(conf, SequenceFile.Writer.file(hfilePartitionFile), SequenceFile.Writer.keyClass(ImmutableBytesWritable.class), SequenceFile.Writer.valueClass(NullWritable.class));
+
+ for (int i = 0; i < splits.size(); i++) {
+ hfilePartitionWriter.append(new ImmutableBytesWritable(splits.get(i)), NullWritable.get());
+ }
+ hfilePartitionWriter.close();
+ }
+
public static void main(String[] args) throws Exception {
int exitCode = ToolRunner.run(new CreateHTableJob(), args);
System.exit(exitCode);
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHFileJob.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHFileJob.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHFileJob.java
index a302daf..b92481e 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHFileJob.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHFileJob.java
@@ -133,6 +133,8 @@ public class CubeHFileJob extends AbstractHadoopJob {
// The reduce tasks should be one more than partition keys
job.setNumReduceTasks(partitionCount+1);
}
+ } else {
+ logger.info("File '" + path.toString() + " doesn't exist, will not reconfigure hfile Partitions");
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionJob.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionJob.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionJob.java
index 36a5732..9018729 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionJob.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionJob.java
@@ -36,6 +36,7 @@ import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
import org.apache.kylin.metadata.model.DataModelDesc;
+import org.apache.kylin.storage.hbase.HBaseConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -95,8 +96,14 @@ public class RangeKeyDistributionJob extends AbstractHadoopJob {
CubeManager cubeMgr = CubeManager.getInstance(KylinConfig.getInstanceFromEnv());
CubeInstance cube = cubeMgr.getCube(cubeName);
KylinConfig config = cube.getConfig();
- int hfileSizeGB = config.getHBaseHFileSizeGB();
- float regionSplitSize = RegionSize.getReionSize(config,cube.getDescriptor());
+ float hfileSizeGB = config.getHBaseHFileSizeGB();
+ float regionSplitSize = config.getKylinHBaseRegionCut();
+
+ int compactionThreshold = Integer.valueOf(HBaseConnection.getCurrentHBaseConfiguration().get("hbase.hstore.compactionThreshold", "3"));
+ if (hfileSizeGB > 0 && hfileSizeGB * compactionThreshold < regionSplitSize) {
+ hfileSizeGB = regionSplitSize / compactionThreshold;
+ logger.info("Adjust hfile size' to " + hfileSizeGB);
+ }
int maxRegionCount = config.getHBaseRegionCountMax();
int minRegionCount = config.getHBaseRegionCountMin();
job.getConfiguration().set(BatchConstants.CFG_OUTPUT_PATH, output.toString());
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionReducer.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionReducer.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionReducer.java
index b694f5f..567dd89 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionReducer.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RangeKeyDistributionReducer.java
@@ -116,21 +116,21 @@ public class RangeKeyDistributionReducer extends KylinReducer<Text, LongWritable
System.out.println(hfilePerRegion + " hfile per region");
Path hfilePartitionFile = new Path(output + "/part-r-00000_hfile");
- try (SequenceFile.Writer hfilePartitionWriter = new SequenceFile.Writer(
+ SequenceFile.Writer hfilePartitionWriter = new SequenceFile.Writer(
hfilePartitionFile.getFileSystem(context.getConfiguration()),
- context.getConfiguration(), hfilePartitionFile, ImmutableBytesWritable.class, NullWritable.class)) {
- int hfileCountInOneRegion = 0;
- for (int i = hfileSizeGB; i < gbPoints.size(); i += hfileSizeGB) {
- hfilePartitionWriter.append(new ImmutableBytesWritable(gbPoints.get(i).getBytes()), NullWritable.get());
- if (++hfileCountInOneRegion >= hfilePerRegion) {
- Text key = gbPoints.get(i);
- outputValue.set(i);
- System.out.println(StringUtils.byteToHexString(key.getBytes()) + "\t" + outputValue.get());
- context.write(key, outputValue);
-
- hfileCountInOneRegion = 0;
- }
+ context.getConfiguration(), hfilePartitionFile, ImmutableBytesWritable.class, NullWritable.class);
+ int hfileCountInOneRegion = 0;
+ for (int i = hfileSizeGB; i < gbPoints.size(); i += hfileSizeGB) {
+ hfilePartitionWriter.append(new ImmutableBytesWritable(gbPoints.get(i).getBytes()), NullWritable.get());
+ if (++hfileCountInOneRegion >= hfilePerRegion) {
+ Text key = gbPoints.get(i);
+ outputValue.set(i);
+ System.out.println(StringUtils.byteToHexString(key.getBytes()) + "\t" + outputValue.get());
+ context.write(key, outputValue);
+
+ hfileCountInOneRegion = 0;
}
}
+ hfilePartitionWriter.close();
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RegionSize.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RegionSize.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RegionSize.java
index 20f3d73..b825b6f 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RegionSize.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RegionSize.java
@@ -29,23 +29,7 @@ public class RegionSize {
protected static final Logger logger = LoggerFactory.getLogger(CreateHTableJob.class);
public static float getReionSize(KylinConfig config, CubeDesc cubeDesc) {
- if (cubeDesc.getRegionSize() != 0) {
- logger.info("Region size specified in Cube desc will be used");
- return cubeDesc.getRegionSize();
- } else {
- logger.info("Region size specified in Model desc will be used");
-
- switch (cubeDesc.getModel().getCapacity().toString()) {
- case "SMALL":
- return config.getKylinHBaseRegionCutSmall();
- case "MEDIUM":
- return config.getKylinHBaseRegionCutMedium();
- case "LARGE":
- return config.getKylinHBaseRegionCutLarge();
- default:
- throw new IllegalArgumentException("Capacity not recognized: " + cubeDesc.getModel().getCapacity().toString());
- }
- }
+ return cubeDesc.getConfig().getKylinHBaseRegionCut();
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/8e7a76b9/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CreateHTableTest.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CreateHTableTest.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CreateHTableTest.java
index 55abe5a..6868533 100644
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CreateHTableTest.java
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CreateHTableTest.java
@@ -25,7 +25,6 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.kylin.common.util.LocalFileMetadataTestCase;
import org.apache.kylin.engine.mr.HadoopUtil;
-import org.apache.kylin.storage.hbase.steps.CreateHTableJob;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -60,7 +59,7 @@ public class CreateHTableTest extends LocalFileMetadataTestCase {
String input = "src/test/resources/partition_list/part-r-00000";
- byte[][] splits = c.getSplits(conf, new Path(input));
+ byte[][] splits = c.getRegionSplits(conf, new Path(input));
assertEquals(497, splits.length);
assertArrayEquals(new byte[] { 0, 0, 0, 0, 0, 0, 15, -1, 11, 51, -45, 2 }, splits[0]);