You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2015/12/25 10:33:49 UTC
[1/4] kylin git commit: KYLIN-1245 Put layer cubing and in-mem cubing
side by side, switch based on 'mapper overlap ratio'
Repository: kylin
Updated Branches:
refs/heads/2.0-rc 8936df133 -> 45d021b73
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseMROutput2.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseMROutput2.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseMROutput2.java
deleted file mode 100644
index e05bf0f..0000000
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseMROutput2.java
+++ /dev/null
@@ -1,290 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
-*/
-
-package org.apache.kylin.storage.hbase.steps;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.commons.lang.StringUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat;
-import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
-import org.apache.hadoop.hbase.mapreduce.TableMapper;
-import org.apache.hadoop.hbase.mapreduce.TableSplit;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.io.SequenceFile;
-import org.apache.hadoop.io.SequenceFile.Reader;
-import org.apache.hadoop.io.SequenceFile.Writer;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.Mapper.Context;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
-import org.apache.kylin.common.util.Pair;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.model.HBaseColumnDesc;
-import org.apache.kylin.cube.model.HBaseColumnFamilyDesc;
-import org.apache.kylin.engine.mr.ByteArrayWritable;
-import org.apache.kylin.engine.mr.HadoopUtil;
-import org.apache.kylin.engine.mr.IMROutput2;
-import org.apache.kylin.job.execution.DefaultChainedExecutable;
-import org.apache.kylin.metadata.model.MeasureDesc;
-
-import com.google.common.collect.Lists;
-
-public class HBaseMROutput2 implements IMROutput2 {
-
- @Override
- public IMRBatchCubingOutputSide2 getBatchCubingOutputSide(final CubeSegment seg) {
- return new IMRBatchCubingOutputSide2() {
- HBaseMRSteps steps = new HBaseMRSteps(seg);
-
- @Override
- public IMRStorageOutputFormat getStorageOutputFormat() {
- return new HBaseOutputFormat(seg);
- }
-
- @Override
- public void addStepPhase2_BuildDictionary(DefaultChainedExecutable jobFlow) {
- jobFlow.addTask(steps.createCreateHTableStepWithStats(jobFlow.getId()));
- }
-
- @Override
- public void addStepPhase3_BuildCube(DefaultChainedExecutable jobFlow) {
- jobFlow.addTask(steps.createBulkLoadStep(jobFlow.getId()));
- }
-
- @Override
- public void addStepPhase4_Cleanup(DefaultChainedExecutable jobFlow) {
- // nothing to do
- }
- };
- }
-
- @Override
- public IMRBatchMergeInputSide2 getBatchMergeInputSide(final CubeSegment seg) {
- return new IMRBatchMergeInputSide2() {
- @Override
- public IMRStorageInputFormat getStorageInputFormat() {
- return new HBaseInputFormat(seg);
- }
- };
- }
-
- @Override
- public IMRBatchMergeOutputSide2 getBatchMergeOutputSide(final CubeSegment seg) {
- return new IMRBatchMergeOutputSide2() {
- HBaseMRSteps steps = new HBaseMRSteps(seg);
-
- @Override
- public IMRStorageOutputFormat getStorageOutputFormat() {
- return new HBaseOutputFormat(seg);
- }
-
- @Override
- public void addStepPhase1_MergeDictionary(DefaultChainedExecutable jobFlow) {
- jobFlow.addTask(steps.createCreateHTableStepWithStats(jobFlow.getId()));
- }
-
- @Override
- public void addStepPhase2_BuildCube(DefaultChainedExecutable jobFlow) {
- jobFlow.addTask(steps.createBulkLoadStep(jobFlow.getId()));
- }
-
- @Override
- public void addStepPhase3_Cleanup(DefaultChainedExecutable jobFlow) {
- jobFlow.addTask(steps.createMergeGCStep());
- }
- };
- }
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- private static class HBaseInputFormat implements IMRStorageInputFormat {
- final CubeSegment seg;
-
- final RowValueDecoder[] rowValueDecoders;
- final ByteArrayWritable parsedKey;
- final Object[] parsedValue;
- final Pair<ByteArrayWritable, Object[]> parsedPair;
-
- public HBaseInputFormat(CubeSegment seg) {
- this.seg = seg;
-
- List<RowValueDecoder> valueDecoderList = Lists.newArrayList();
- List<MeasureDesc> measuresDescs = Lists.newArrayList();
- for (HBaseColumnFamilyDesc cfDesc : seg.getCubeDesc().getHBaseMapping().getColumnFamily()) {
- for (HBaseColumnDesc colDesc : cfDesc.getColumns()) {
- valueDecoderList.add(new RowValueDecoder(colDesc));
- for (MeasureDesc measure : colDesc.getMeasures()) {
- measuresDescs.add(measure);
- }
- }
- }
- this.rowValueDecoders = valueDecoderList.toArray(new RowValueDecoder[valueDecoderList.size()]);
-
- this.parsedKey = new ByteArrayWritable();
- this.parsedValue = new Object[measuresDescs.size()];
- this.parsedPair = new Pair<ByteArrayWritable, Object[]>(parsedKey, parsedValue);
- }
-
- @Override
- public void configureInput(Class<? extends Mapper> mapperClz, Class<? extends WritableComparable> outputKeyClz, Class<? extends Writable> outputValueClz, Job job) throws IOException {
- Configuration conf = job.getConfiguration();
- HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf));
-
- List<Scan> scans = new ArrayList<Scan>();
- for (String htable : new HBaseMRSteps(seg).getMergingHTables()) {
- Scan scan = new Scan();
- scan.setCaching(512); // 1 is the default in Scan, which will be bad for MapReduce jobs
- scan.setCacheBlocks(false); // don't set to true for MR jobs
- scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(htable));
- scans.add(scan);
- }
-
- TableMapReduceUtil.initTableMapperJob(scans, (Class<? extends TableMapper>) mapperClz, outputKeyClz, outputValueClz, job);
- TableMapReduceUtil.initCredentials(job);
-
- }
-
- @Override
- public CubeSegment findSourceSegment(Context context) throws IOException {
- TableSplit currentSplit = (TableSplit) context.getInputSplit();
- byte[] tableName = currentSplit.getTableName();
- String htableName = Bytes.toString(tableName);
-
- // decide which source segment
- for (CubeSegment segment : seg.getCubeInstance().getSegments()) {
- String segmentHtable = segment.getStorageLocationIdentifier();
- if (segmentHtable != null && segmentHtable.equalsIgnoreCase(htableName)) {
- return segment;
- }
- }
- throw new IllegalStateException("No merging segment's storage location identifier equals " + htableName);
- }
-
- @Override
- public Pair<ByteArrayWritable, Object[]> parseMapperInput(Object inKey, Object inValue) {
- ImmutableBytesWritable key = (ImmutableBytesWritable) inKey;
- parsedKey.set(key.get(), key.getOffset(), key.getLength());
-
- Result hbaseRow = (Result) inValue;
- for (int i = 0; i < rowValueDecoders.length; i++) {
- rowValueDecoders[i].decode(hbaseRow);
- rowValueDecoders[i].loadCubeMeasureArray(parsedValue);
- }
-
- return parsedPair;
- }
- }
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- private static class HBaseOutputFormat implements IMRStorageOutputFormat {
- final CubeSegment seg;
-
- final List<KeyValueCreator> keyValueCreators = Lists.newArrayList();
- final ImmutableBytesWritable outputKey = new ImmutableBytesWritable();
-
- public HBaseOutputFormat(CubeSegment seg) {
- this.seg = seg;
- }
-
- @Override
- public void configureOutput(Class<? extends Reducer> reducer, String jobFlowId, Job job) throws IOException {
- Path hfilePath = new Path(new HBaseMRSteps(seg).getHFilePath(jobFlowId));
- FileOutputFormat.setOutputPath(job, hfilePath);
-
- String htableName = seg.getStorageLocationIdentifier();
- Configuration conf = HBaseConfiguration.create(job.getConfiguration());
- HTable htable = new HTable(conf, htableName);
- HFileOutputFormat.configureIncrementalLoad(job, htable);
-
- // set Reducer; This need be after configureIncrementalLoad, to overwrite the default reducer class
- job.setReducerClass(reducer);
-
- // kylin uses ByteArrayWritable instead of ImmutableBytesWritable as mapper output key
- rewriteTotalOrderPartitionerFile(job);
-
- HadoopUtil.deletePath(job.getConfiguration(), hfilePath);
- }
-
- @Override
- public void doReducerOutput(ByteArrayWritable key, Object[] value, Reducer.Context context) throws IOException, InterruptedException {
- if (keyValueCreators.size() == 0) {
- for (HBaseColumnFamilyDesc cfDesc : seg.getCubeDesc().getHBaseMapping().getColumnFamily()) {
- for (HBaseColumnDesc colDesc : cfDesc.getColumns()) {
- keyValueCreators.add(new KeyValueCreator(seg.getCubeDesc(), colDesc));
- }
- }
- }
-
- outputKey.set(key.array(), key.offset(), key.length());
-
- KeyValue outputValue;
- for (int i = 0; i < keyValueCreators.size(); i++) {
- outputValue = keyValueCreators.get(i).create(key.array(), key.offset(), key.length(), value);
- context.write(outputKey, outputValue);
- }
- }
-
- private void rewriteTotalOrderPartitionerFile(Job job) throws IOException {
- Configuration conf = job.getConfiguration();
- String partitionsFile = TotalOrderPartitioner.getPartitionFile(conf);
- if (StringUtils.isBlank(partitionsFile))
- throw new IllegalStateException("HFileOutputFormat.configureIncrementalLoad don't configure TotalOrderPartitioner any more?");
-
- Path partitionsPath = new Path(partitionsFile);
-
- // read in partition file in ImmutableBytesWritable
- List<ByteArrayWritable> keys = Lists.newArrayList();
- Reader reader = new SequenceFile.Reader(conf, Reader.file(partitionsPath));
- try {
- ImmutableBytesWritable key = new ImmutableBytesWritable();
- while (reader.next(key, NullWritable.get())) {
- keys.add(new ByteArrayWritable(key.copyBytes()));
- }
- } finally {
- reader.close();
- }
-
- // write out again in ByteArrayWritable
- Writer writer = SequenceFile.createWriter(conf, Writer.file(partitionsPath), Writer.keyClass(ByteArrayWritable.class), Writer.valueClass(NullWritable.class));
- try {
- for (ByteArrayWritable key : keys) {
- writer.append(key, NullWritable.get());
- }
- } finally {
- writer.close();
- }
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseMROutput2Transition.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseMROutput2Transition.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseMROutput2Transition.java
index df88dc1..9036cf4 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseMROutput2Transition.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HBaseMROutput2Transition.java
@@ -18,53 +18,12 @@
package org.apache.kylin.storage.hbase.steps;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
-import org.apache.hadoop.hbase.mapreduce.TableMapper;
-import org.apache.hadoop.hbase.mapreduce.TableSplit;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.Mapper.Context;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mapreduce.lib.input.FileSplit;
-import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.util.Pair;
import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.kv.RowConstants;
-import org.apache.kylin.cube.model.HBaseColumnDesc;
-import org.apache.kylin.cube.model.HBaseColumnFamilyDesc;
-import org.apache.kylin.engine.mr.ByteArrayWritable;
-import org.apache.kylin.engine.mr.HadoopUtil;
import org.apache.kylin.engine.mr.IMROutput2;
-import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
-import org.apache.kylin.engine.mr.steps.MergeCuboidMapper;
import org.apache.kylin.job.execution.DefaultChainedExecutable;
-import org.apache.kylin.metadata.measure.MeasureCodec;
-import org.apache.kylin.metadata.model.MeasureDesc;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.Lists;
-
/**
* This "Transition" impl generates cuboid files and then convert to HFile.
* The additional step slows down build process, but the gains is merge
@@ -77,6 +36,7 @@ import com.google.common.collect.Lists;
*/
public class HBaseMROutput2Transition implements IMROutput2 {
+ @SuppressWarnings("unused")
private static final Logger logger = LoggerFactory.getLogger(HBaseMROutput2Transition.class);
@Override
@@ -85,19 +45,12 @@ public class HBaseMROutput2Transition implements IMROutput2 {
HBaseMRSteps steps = new HBaseMRSteps(seg);
@Override
- public IMRStorageOutputFormat getStorageOutputFormat() {
- return new HBaseOutputFormat(seg);
- }
-
- @Override
public void addStepPhase2_BuildDictionary(DefaultChainedExecutable jobFlow) {
jobFlow.addTask(steps.createCreateHTableStepWithStats(jobFlow.getId()));
}
@Override
- public void addStepPhase3_BuildCube(DefaultChainedExecutable jobFlow) {
- String cuboidRootPath = steps.getCuboidRootPath(jobFlow.getId());
-
+ public void addStepPhase3_BuildCube(DefaultChainedExecutable jobFlow, String cuboidRootPath) {
jobFlow.addTask(steps.createConvertCuboidToHfileStep(cuboidRootPath, jobFlow.getId()));
jobFlow.addTask(steps.createBulkLoadStep(jobFlow.getId()));
}
@@ -110,34 +63,17 @@ public class HBaseMROutput2Transition implements IMROutput2 {
}
@Override
- public IMRBatchMergeInputSide2 getBatchMergeInputSide(final CubeSegment seg) {
- return new IMRBatchMergeInputSide2() {
- @Override
- public IMRStorageInputFormat getStorageInputFormat() {
- return new HBaseInputFormat(seg);
- }
- };
- }
-
- @Override
public IMRBatchMergeOutputSide2 getBatchMergeOutputSide(final CubeSegment seg) {
return new IMRBatchMergeOutputSide2() {
HBaseMRSteps steps = new HBaseMRSteps(seg);
@Override
- public IMRStorageOutputFormat getStorageOutputFormat() {
- return new HBaseOutputFormat(seg);
- }
-
- @Override
public void addStepPhase1_MergeDictionary(DefaultChainedExecutable jobFlow) {
jobFlow.addTask(steps.createCreateHTableStepWithStats(jobFlow.getId()));
}
@Override
- public void addStepPhase2_BuildCube(DefaultChainedExecutable jobFlow) {
- String cuboidRootPath = steps.getCuboidRootPath(jobFlow.getId());
-
+ public void addStepPhase2_BuildCube(DefaultChainedExecutable jobFlow, String cuboidRootPath) {
jobFlow.addTask(steps.createConvertCuboidToHfileStep(cuboidRootPath, jobFlow.getId()));
jobFlow.addTask(steps.createBulkLoadStep(jobFlow.getId()));
}
@@ -148,263 +84,4 @@ public class HBaseMROutput2Transition implements IMROutput2 {
}
};
}
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- private static class HBaseInputFormat implements IMRStorageInputFormat {
- final CubeSegment seg;
-
- public HBaseInputFormat(CubeSegment seg) {
- this.seg = seg;
- }
-
- @Override
- public void configureInput(Class<? extends Mapper> mapperClz, Class<? extends WritableComparable> outputKeyClz, Class<? extends Writable> outputValueClz, Job job) throws IOException {
- // merge by cuboid files
- if (isMergeFromCuboidFiles(job.getConfiguration())) {
- logger.info("Merge from cuboid files for " + seg);
-
- job.setInputFormatClass(SequenceFileInputFormat.class);
- addCuboidInputDirs(job);
-
- job.setMapperClass(mapperClz);
- job.setMapOutputKeyClass(outputKeyClz);
- job.setMapOutputValueClass(outputValueClz);
- }
- // merge from HTable scan
- else {
- logger.info("Merge from HTables for " + seg);
-
- Configuration conf = job.getConfiguration();
- HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf));
-
- List<Scan> scans = new ArrayList<Scan>();
- for (String htable : new HBaseMRSteps(seg).getMergingHTables()) {
- Scan scan = new Scan();
- scan.setCaching(512); // 1 is the default in Scan, which will be bad for MapReduce jobs
- scan.setCacheBlocks(false); // don't set to true for MR jobs
- scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(htable));
- scans.add(scan);
- }
- TableMapReduceUtil.initTableMapperJob(scans, (Class<? extends TableMapper>) mapperClz, outputKeyClz, outputValueClz, job);
- TableMapReduceUtil.initCredentials(job);
- }
- }
-
- private void addCuboidInputDirs(Job job) throws IOException {
- List<CubeSegment> mergingSegments = seg.getCubeInstance().getMergingSegments(seg);
- HBaseMRSteps steps = new HBaseMRSteps(seg);
-
- String[] inputs = new String[mergingSegments.size()];
- for (int i = 0; i < mergingSegments.size(); i++) {
- CubeSegment mergingSeg = mergingSegments.get(i);
- String cuboidPath = steps.getCuboidRootPath(mergingSeg);
- inputs[i] = cuboidPath + (cuboidPath.endsWith("/") ? "" : "/") + "*";
- }
-
- AbstractHadoopJob.addInputDirs(inputs, job);
- }
-
- @Override
- public CubeSegment findSourceSegment(Context context) throws IOException {
- // merge by cuboid files
- if (isMergeFromCuboidFiles(context.getConfiguration())) {
- FileSplit fileSplit = (FileSplit) context.getInputSplit();
- return MergeCuboidMapper.findSourceSegment(fileSplit, seg.getCubeInstance());
- }
- // merge from HTable scan
- else {
- TableSplit currentSplit = (TableSplit) context.getInputSplit();
- byte[] tableName = currentSplit.getTableName();
- String htableName = Bytes.toString(tableName);
-
- // decide which source segment
- for (CubeSegment segment : seg.getCubeInstance().getSegments()) {
- String segmentHtable = segment.getStorageLocationIdentifier();
- if (segmentHtable != null && segmentHtable.equalsIgnoreCase(htableName)) {
- return segment;
- }
- }
- throw new IllegalStateException("No merging segment's storage location identifier equals " + htableName);
- }
- }
-
- transient ByteArrayWritable parsedKey;
- transient Object[] parsedValue;
- transient Pair<ByteArrayWritable, Object[]> parsedPair;
-
- transient MeasureCodec measureCodec;
- transient RowValueDecoder[] rowValueDecoders;
-
- @Override
- public Pair<ByteArrayWritable, Object[]> parseMapperInput(Object inKey, Object inValue) {
- // lazy init
- if (parsedPair == null) {
- parsedKey = new ByteArrayWritable();
- parsedValue = new Object[seg.getCubeDesc().getMeasures().size()];
- parsedPair = new Pair<ByteArrayWritable, Object[]>(parsedKey, parsedValue);
- }
-
- // merge by cuboid files
- if (isMergeFromCuboidFiles(null)) {
- return parseMapperInputFromCuboidFile(inKey, inValue);
- }
- // merge from HTable scan
- else {
- return parseMapperInputFromHTable(inKey, inValue);
- }
- }
-
- private Pair<ByteArrayWritable, Object[]> parseMapperInputFromCuboidFile(Object inKey, Object inValue) {
- // lazy init
- if (measureCodec == null) {
- measureCodec = new MeasureCodec(seg.getCubeDesc().getMeasures());
- }
-
- Text key = (Text) inKey;
- parsedKey.set(key.getBytes(), 0, key.getLength());
-
- Text value = (Text) inValue;
- measureCodec.decode(ByteBuffer.wrap(value.getBytes(), 0, value.getLength()), parsedValue);
-
- return parsedPair;
- }
-
- private Pair<ByteArrayWritable, Object[]> parseMapperInputFromHTable(Object inKey, Object inValue) {
- // lazy init
- if (rowValueDecoders == null) {
- List<RowValueDecoder> valueDecoderList = Lists.newArrayList();
- List<MeasureDesc> measuresDescs = Lists.newArrayList();
- for (HBaseColumnFamilyDesc cfDesc : seg.getCubeDesc().getHBaseMapping().getColumnFamily()) {
- for (HBaseColumnDesc colDesc : cfDesc.getColumns()) {
- valueDecoderList.add(new RowValueDecoder(colDesc));
- for (MeasureDesc measure : colDesc.getMeasures()) {
- measuresDescs.add(measure);
- }
- }
- }
- rowValueDecoders = valueDecoderList.toArray(new RowValueDecoder[valueDecoderList.size()]);
- }
-
- ImmutableBytesWritable key = (ImmutableBytesWritable) inKey;
- parsedKey.set(key.get(), key.getOffset(), key.getLength());
-
- Result hbaseRow = (Result) inValue;
- for (int i = 0; i < rowValueDecoders.length; i++) {
- rowValueDecoders[i].decode(hbaseRow);
- rowValueDecoders[i].loadCubeMeasureArray(parsedValue);
- }
-
- return parsedPair;
- }
-
- transient Boolean isMergeFromCuboidFiles;
-
- // merge from cuboid files is better than merge from HTable, because no workload on HBase region server
- private boolean isMergeFromCuboidFiles(Configuration jobConf) {
- // cache in this object?
- if (isMergeFromCuboidFiles != null)
- return isMergeFromCuboidFiles.booleanValue();
-
- final String confKey = "kylin.isMergeFromCuboidFiles";
-
- // cache in job configuration?
- if (jobConf != null) {
- String result = jobConf.get(confKey);
- if (result != null) {
- isMergeFromCuboidFiles = Boolean.valueOf(result);
- return isMergeFromCuboidFiles.booleanValue();
- }
- }
-
- boolean result = true;
-
- try {
- List<CubeSegment> mergingSegments = seg.getCubeInstance().getMergingSegments(seg);
- HBaseMRSteps steps = new HBaseMRSteps(seg);
- for (CubeSegment mergingSeg : mergingSegments) {
- String cuboidRootPath = steps.getCuboidRootPath(mergingSeg);
- FileSystem fs = HadoopUtil.getFileSystem(cuboidRootPath);
-
- boolean cuboidFileExist = fs.exists(new Path(cuboidRootPath));
- if (cuboidFileExist == false) {
- logger.info("Merge from HTable because " + cuboidRootPath + " does not exist");
- result = false;
- break;
- }
- }
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
-
- // put in cache
- isMergeFromCuboidFiles = Boolean.valueOf(result);
- if (jobConf != null) {
- jobConf.set(confKey, "" + result);
- }
- return result;
- }
- }
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- private static class HBaseOutputFormat implements IMRStorageOutputFormat {
- final CubeSegment seg;
-
- Text outputKey;
- Text outputValue;
- ByteBuffer valueBuf;
- MeasureCodec codec;
-
- public HBaseOutputFormat(CubeSegment seg) {
- this.seg = seg;
- }
-
- @Override
- public void configureOutput(Class<? extends Reducer> reducer, String jobFlowId, Job job) throws IOException {
- job.setReducerClass(reducer);
-
- // estimate the reducer number
- String htableName = seg.getStorageLocationIdentifier();
- Configuration conf = HBaseConfiguration.create(job.getConfiguration());
- HTable htable = new HTable(conf, htableName);
- int regions = htable.getStartKeys().length + 1;
- htable.close();
-
- int reducerNum = regions * 3;
- reducerNum = Math.max(1, reducerNum);
- KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
- reducerNum = Math.min(kylinConfig.getHadoopJobMaxReducerNumber(), reducerNum);
-
- job.setNumReduceTasks(reducerNum);
-
- // the cuboid file and KV class must be compatible with 0.7 version for smooth upgrade
- job.setOutputFormatClass(SequenceFileOutputFormat.class);
- job.setOutputKeyClass(Text.class);
- job.setOutputValueClass(Text.class);
-
- Path output = new Path(new HBaseMRSteps(seg).getCuboidRootPath(jobFlowId));
- FileOutputFormat.setOutputPath(job, output);
-
- HadoopUtil.deletePath(job.getConfiguration(), output);
- }
-
- @Override
- public void doReducerOutput(ByteArrayWritable key, Object[] value, Reducer.Context context) throws IOException, InterruptedException {
- // lazy init
- if (outputKey == null) {
- outputKey = new Text();
- outputValue = new Text();
- valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
- codec = new MeasureCodec(seg.getCubeDesc().getMeasures());
- }
-
- outputKey.set(key.array(), key.offset(), key.length());
-
- valueBuf.clear();
- codec.encode(value, valueBuf);
- outputValue.set(valueBuf.array(), 0, valueBuf.position());
-
- context.write(outputKey, outputValue);
- }
- }
-
}
[4/4] kylin git commit: KYLIN-1245 fix compile error after code merge
Posted by li...@apache.org.
KYLIN-1245 fix compile error after code merge
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/45d021b7
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/45d021b7
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/45d021b7
Branch: refs/heads/2.0-rc
Commit: 45d021b730bf744e5b0ada9fcdcd5f21f4a24474
Parents: 2aedddf
Author: Li, Yang <ya...@ebay.com>
Authored: Fri Dec 25 17:33:37 2015 +0800
Committer: Li, Yang <ya...@ebay.com>
Committed: Fri Dec 25 17:33:37 2015 +0800
----------------------------------------------------------------------
.../cube/model/validation/rule/AggregationGroupSizeRule.java | 2 +-
.../apache/kylin/cube/model/validation/rule/FunctionRule.java | 2 +-
.../java/org/apache/kylin/engine/mr/BatchCubingJobBuilder2.java | 2 +-
.../java/org/apache/kylin/engine/mr/common/CubeStatsReader.java | 4 ++--
4 files changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/45d021b7/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/AggregationGroupSizeRule.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/AggregationGroupSizeRule.java b/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/AggregationGroupSizeRule.java
index e37b9a9..e42edc1 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/AggregationGroupSizeRule.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/AggregationGroupSizeRule.java
@@ -60,7 +60,7 @@ public class AggregationGroupSizeRule implements IValidatorRule<CubeDesc> {
}
protected int getMaxAgrGroupSize() {
- String size = KylinConfig.getInstanceFromEnv().getProperty(KEY_MAX_AGR_GROUP_SIZE, String.valueOf(DEFAULT_MAX_AGR_GROUP_SIZE));
+ String size = KylinConfig.getInstanceFromEnv().getOptional(KEY_MAX_AGR_GROUP_SIZE, String.valueOf(DEFAULT_MAX_AGR_GROUP_SIZE));
return Integer.parseInt(size);
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/45d021b7/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/FunctionRule.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/FunctionRule.java b/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/FunctionRule.java
index 1920fc7..881ef44 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/FunctionRule.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/validation/rule/FunctionRule.java
@@ -123,7 +123,7 @@ public class FunctionRule implements IValidatorRule<CubeDesc> {
context.addResult(ResultLevel.ERROR, "Return type for function " + func + " must be one of " + DataType.NUMBER_FAMILY);
}
} else {
- if (StringUtils.equalsIgnoreCase(KylinConfig.getInstanceFromEnv().getProperty(KEY_IGNORE_UNKNOWN_FUNC, "false"), "false")) {
+ if (StringUtils.equalsIgnoreCase(KylinConfig.getInstanceFromEnv().getOptional(KEY_IGNORE_UNKNOWN_FUNC, "false"), "false")) {
context.addResult(ResultLevel.ERROR, "Unrecognized function: [" + func + "]");
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/45d021b7/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder2.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder2.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder2.java
index 86439a8..292fa1d 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder2.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder2.java
@@ -74,7 +74,7 @@ public class BatchCubingJobBuilder2 extends JobBuilderSupport {
private void addLayerCubingSteps(final CubingJob result, final String jobId, final String cuboidRootPath) {
RowKeyDesc rowKeyDesc = ((CubeSegment) seg).getCubeDesc().getRowkey();
- final int groupRowkeyColumnsCount = ((CubeSegment) seg).getCubeDesc().getBuildLevel();
+ final int groupRowkeyColumnsCount = rowKeyDesc.getNCuboidBuildLevels();
final int totalRowkeyColumnsCount = rowKeyDesc.getRowKeyColumns().length;
final String[] cuboidOutputTempPath = getCuboidOutputPaths(cuboidRootPath, totalRowkeyColumnsCount, groupRowkeyColumnsCount);
// base cuboid step
http://git-wip-us.apache.org/repos/asf/kylin/blob/45d021b7/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java
index 58900e0..10a4a9c 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java
@@ -51,7 +51,7 @@ import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.engine.mr.HadoopUtil;
import org.apache.kylin.engine.mr.steps.InMemCuboidJob;
-import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.model.DataType;
import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
import org.apache.kylin.metadata.model.TblColRef;
@@ -211,7 +211,7 @@ public class CubeStatsReader {
int space = 0;
boolean isMemoryHungry = false;
for (MeasureDesc measureDesc : cubeSegment.getCubeDesc().getMeasures()) {
- if (measureDesc.getFunction().getMeasureType().isMemoryHungry()) {
+ if (measureDesc.getFunction().isCountDistinct()) {
isMemoryHungry = true;
}
DataType returnType = measureDesc.getFunction().getReturnDataType();
[2/4] kylin git commit: KYLIN-1245 Put layer cubing and in-mem cubing
side by side, switch based on 'mapper overlap ratio'
Posted by li...@apache.org.
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java
new file mode 100644
index 0000000..58900e0
--- /dev/null
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CubeStatsReader.java
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.kylin.engine.mr.common;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PrintWriter;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import javax.annotation.Nullable;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.SequenceFile.Reader;
+import org.apache.hadoop.io.SequenceFile.Reader.Option;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.hll.HyperLogLogPlusCounter;
+import org.apache.kylin.common.persistence.ResourceStore;
+import org.apache.kylin.common.util.ByteArray;
+import org.apache.kylin.common.util.Bytes;
+import org.apache.kylin.cube.CubeInstance;
+import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.cube.model.CubeDesc;
+import org.apache.kylin.engine.mr.HadoopUtil;
+import org.apache.kylin.engine.mr.steps.InMemCuboidJob;
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.model.MeasureDesc;
+import org.apache.kylin.metadata.model.SegmentStatusEnum;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+/**
+ * This should be in cube module. It's here in engine-mr because currently stats
+ * are saved as sequence files thus a hadoop dependency.
+ */
+public class CubeStatsReader {
+
+ private static final Logger logger = LoggerFactory.getLogger(InMemCuboidJob.class);
+
+ final CubeSegment seg;
+ final int samplingPercentage;
+ final double mapperOverlapRatioOfFirstBuild; // only makes sense for the first build, is meaningless after merge
+ final Map<Long, HyperLogLogPlusCounter> cuboidRowCountMap;
+
+ public CubeStatsReader(CubeSegment cubeSegment, KylinConfig kylinConfig) throws IOException {
+ ResourceStore store = ResourceStore.getStore(kylinConfig);
+ String statsKey = cubeSegment.getStatisticsResourcePath();
+ File tmpSeqFile = writeTmpSeqFile(store.getResource(statsKey).inputStream);
+ Reader reader = null;
+
+ try {
+ Configuration hadoopConf = HadoopUtil.getCurrentConfiguration();
+
+ Option seqInput = SequenceFile.Reader.file(new Path("file://" + tmpSeqFile.getAbsolutePath()));
+ reader = new SequenceFile.Reader(hadoopConf, seqInput);
+
+ int percentage = 100;
+ double mapperOverlapRatio = 0;
+ Map<Long, HyperLogLogPlusCounter> counterMap = Maps.newHashMap();
+
+ LongWritable key = (LongWritable) ReflectionUtils.newInstance(reader.getKeyClass(), hadoopConf);
+ BytesWritable value = (BytesWritable) ReflectionUtils.newInstance(reader.getValueClass(), hadoopConf);
+ while (reader.next(key, value)) {
+ if (key.get() == 0L) {
+ percentage = Bytes.toInt(value.getBytes());
+ } else if (key.get() == -1) {
+ mapperOverlapRatio = Bytes.toDouble(value.getBytes());
+ } else {
+ HyperLogLogPlusCounter hll = new HyperLogLogPlusCounter(14);
+ ByteArray byteArray = new ByteArray(value.getBytes());
+ hll.readRegisters(byteArray.asBuffer());
+ counterMap.put(key.get(), hll);
+ }
+ }
+
+ this.seg = cubeSegment;
+ this.samplingPercentage = percentage;
+ this.mapperOverlapRatioOfFirstBuild = mapperOverlapRatio;
+ this.cuboidRowCountMap = counterMap;
+
+ } finally {
+ IOUtils.closeStream(reader);
+ tmpSeqFile.delete();
+ }
+ }
+
+ private File writeTmpSeqFile(InputStream inputStream) throws IOException {
+ File tempFile = File.createTempFile("kylin_stats_tmp", ".seq");
+ FileOutputStream out = null;
+ try {
+ out = new FileOutputStream(tempFile);
+ org.apache.commons.io.IOUtils.copy(inputStream, out);
+ } finally {
+ IOUtils.closeStream(inputStream);
+ IOUtils.closeStream(out);
+ }
+ return tempFile;
+ }
+
+ public Map<Long, Long> getCuboidRowCountMap() {
+ return getCuboidRowCountMapFromSampling(cuboidRowCountMap, samplingPercentage);
+ }
+
+ // return map of Cuboid ID => MB
+ public Map<Long, Double> getCuboidSizeMap() {
+ return getCuboidSizeMapFromRowCount(seg, getCuboidRowCountMap());
+ }
+
+ public double getMapperOverlapRatioOfFirstBuild() {
+ return mapperOverlapRatioOfFirstBuild;
+ }
+
+ public void print(PrintWriter out) {
+ Map<Long, Long> rowCountMap = getCuboidRowCountMap();
+ Map<Long, Double> sizeMap = getCuboidSizeMap();
+ List<Long> cuboids = new ArrayList<Long>(rowCountMap.keySet());
+ Collections.sort(cuboids);
+
+ out.println("============================================================================");
+ out.println("Statistics of " + seg);
+ out.println(" Sampling percentage: " + samplingPercentage);
+ out.println(" Mapper overlap ratio: " + mapperOverlapRatioOfFirstBuild);
+ for (Long cuboid : cuboids) {
+ out.println(" Cuboid :\t" + rowCountMap.get(cuboid) + " rows, " + sizeMap.get(cuboid) + " MB");
+ }
+ out.println("----------------------------------------------------------------------------");
+ }
+
+ public static Map<Long, Long> getCuboidRowCountMapFromSampling(Map<Long, HyperLogLogPlusCounter> hllcMap, int samplingPercentage) {
+ return Maps.transformValues(hllcMap, new Function<HyperLogLogPlusCounter, Long>() {
+ @Nullable
+ @Override
+ public Long apply(HyperLogLogPlusCounter input) {
+ // No need to adjust according sampling percentage. Assumption is that data set is far
+ // more than cardinality. Even a percentage of the data should already see all cardinalities.
+ return input.getCountEstimate();
+ }
+ });
+ }
+
+ public static Map<Long, Double> getCuboidSizeMapFromRowCount(CubeSegment cubeSegment, Map<Long, Long> rowCountMap) {
+ final CubeDesc cubeDesc = cubeSegment.getCubeDesc();
+ final List<Integer> rowkeyColumnSize = Lists.newArrayList();
+ final long baseCuboidId = Cuboid.getBaseCuboidId(cubeDesc);
+ final Cuboid baseCuboid = Cuboid.findById(cubeDesc, baseCuboidId);
+ final List<TblColRef> columnList = baseCuboid.getColumns();
+
+ for (int i = 0; i < columnList.size(); i++) {
+ rowkeyColumnSize.add(cubeSegment.getColumnLength(columnList.get(i)));
+ }
+
+ Map<Long, Double> sizeMap = Maps.newHashMap();
+ for (Map.Entry<Long, Long> entry : rowCountMap.entrySet()) {
+ sizeMap.put(entry.getKey(), estimateCuboidStorageSize(cubeSegment, entry.getKey(), entry.getValue(), baseCuboidId, rowkeyColumnSize));
+ }
+ return sizeMap;
+ }
+
+ /**
+ * Estimate the cuboid's size
+ *
+ * @return the cuboid size in M bytes
+ */
+ private static double estimateCuboidStorageSize(CubeSegment cubeSegment, long cuboidId, long rowCount, long baseCuboidId, List<Integer> rowKeyColumnLength) {
+
+ int bytesLength = cubeSegment.getRowKeyPreambleSize();
+
+ long mask = Long.highestOneBit(baseCuboidId);
+ long parentCuboidIdActualLength = Long.SIZE - Long.numberOfLeadingZeros(baseCuboidId);
+ for (int i = 0; i < parentCuboidIdActualLength; i++) {
+ if ((mask & cuboidId) > 0) {
+ bytesLength += rowKeyColumnLength.get(i); //colIO.getColumnLength(columnList.get(i));
+ }
+ mask = mask >> 1;
+ }
+
+ // add the measure length
+ int space = 0;
+ boolean isMemoryHungry = false;
+ for (MeasureDesc measureDesc : cubeSegment.getCubeDesc().getMeasures()) {
+ if (measureDesc.getFunction().getMeasureType().isMemoryHungry()) {
+ isMemoryHungry = true;
+ }
+ DataType returnType = measureDesc.getFunction().getReturnDataType();
+ space += returnType.getStorageBytesEstimate();
+ }
+ bytesLength += space;
+
+ double ret = 1.0 * bytesLength * rowCount / (1024L * 1024L);
+ if (isMemoryHungry) {
+ logger.info("Cube is memory hungry, storage size estimation multiply 0.05");
+ ret *= 0.05;
+ } else {
+ logger.info("Cube is not memory hungry, storage size estimation multiply 0.25");
+ ret *= 0.25;
+ }
+ logger.info("Cuboid " + cuboidId + " has " + rowCount + " rows, each row size is " + bytesLength + " bytes." + " Total size is " + ret + "M.");
+ return ret;
+ }
+
+ public static void main(String[] args) throws IOException {
+ KylinConfig config = KylinConfig.getInstanceFromEnv();
+ CubeInstance cube = CubeManager.getInstance(config).getCube(args[0]);
+ List<CubeSegment> segments = cube.getSegments(SegmentStatusEnum.READY);
+
+ PrintWriter out = new PrintWriter(System.out);
+ for (CubeSegment seg : segments) {
+ new CubeStatsReader(seg, config).print(out);
+ }
+ out.flush();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CuboidStatsUtil.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CuboidStatsUtil.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CuboidStatsUtil.java
index 10c82c3..2596a93 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CuboidStatsUtil.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CuboidStatsUtil.java
@@ -36,7 +36,13 @@ import org.apache.kylin.cube.kv.RowConstants;
public class CuboidStatsUtil {
- public static void writeCuboidStatistics(Configuration conf, Path outputPath, Map<Long, HyperLogLogPlusCounter> cuboidHLLMap, int samplingPercentage) throws IOException {
+ public static void writeCuboidStatistics(Configuration conf, Path outputPath, //
+ Map<Long, HyperLogLogPlusCounter> cuboidHLLMap, int samplingPercentage) throws IOException {
+ writeCuboidStatistics(conf, outputPath, cuboidHLLMap, samplingPercentage, 0);
+ }
+
+ public static void writeCuboidStatistics(Configuration conf, Path outputPath, //
+ Map<Long, HyperLogLogPlusCounter> cuboidHLLMap, int samplingPercentage, double mapperOverlapRatio) throws IOException {
Path seqFilePath = new Path(outputPath, BatchConstants.CFG_STATISTICS_CUBOID_ESTIMATION);
SequenceFile.Writer writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(seqFilePath), SequenceFile.Writer.keyClass(LongWritable.class), SequenceFile.Writer.valueClass(BytesWritable.class));
@@ -48,6 +54,12 @@ public class CuboidStatsUtil {
writer.append(new LongWritable(0l), new BytesWritable(Bytes.toBytes(samplingPercentage)));
ByteBuffer valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
try {
+ // mapper overlap ratio at key -1
+ writer.append(new LongWritable(-1), new BytesWritable(Bytes.toBytes(mapperOverlapRatio)));
+
+ // sampling percentage at key 0
+ writer.append(new LongWritable(0l), new BytesWritable(Bytes.toBytes(samplingPercentage)));
+
for (long i : allCuboids) {
valueBuf.clear();
cuboidHLLMap.get(i).writeRegisters(valueBuf);
@@ -58,4 +70,5 @@ public class CuboidStatsUtil {
writer.close();
}
}
+
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/MapReduceExecutable.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/MapReduceExecutable.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/MapReduceExecutable.java
index 46557c8..1449fa4 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/MapReduceExecutable.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/MapReduceExecutable.java
@@ -117,6 +117,10 @@ public class MapReduceExecutable extends AbstractExecutable {
try {
//for async mr job, ToolRunner just return 0;
ToolRunner.run(hadoopJob, args);
+
+ if (hadoopJob.isSkipped()) {
+ return new ExecuteResult(ExecuteResult.State.SUCCEED, "skipped");
+ }
} catch (Exception ex) {
StringBuilder log = new StringBuilder();
logger.error("error execute " + this.toString(), ex);
@@ -203,7 +207,7 @@ public class MapReduceExecutable extends AbstractExecutable {
if (yarnStatusCheckUrl != null) {
return yarnStatusCheckUrl;
} else {
- logger.info(KylinConfig.KYLIN_JOB_YARN_APP_REST_CHECK_URL + " is not set, read from job configuration");
+ logger.info("kylin.job.yarn.app.rest.check.status.url" + " is not set, read from job configuration");
}
String rmWebHost = HAUtil.getConfValueForRMInstance(YarnConfiguration.RM_WEBAPP_ADDRESS, YarnConfiguration.DEFAULT_RM_WEBAPP_ADDRESS, job.getConfiguration());
if(HAUtil.isHAEnabled(job.getConfiguration())) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidJob.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidJob.java
index e7341bf..8ff784d 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidJob.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidJob.java
@@ -26,10 +26,11 @@ import org.apache.hadoop.util.ToolRunner;
*/
public class BaseCuboidJob extends CuboidJob {
+
public BaseCuboidJob() {
this.setMapperClass(HiveToBaseCuboidMapper.class);
}
-
+
public static void main(String[] args) throws Exception {
CuboidJob job = new BaseCuboidJob();
int exitCode = ToolRunner.run(job, args);
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CuboidJob.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CuboidJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CuboidJob.java
index 3edaefb..1647480 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CuboidJob.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CuboidJob.java
@@ -37,11 +37,13 @@ import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.CuboidCLI;
import org.apache.kylin.cube.model.CubeDesc;
+import org.apache.kylin.engine.mr.CubingJob;
import org.apache.kylin.engine.mr.IMRInput.IMRTableInputFormat;
import org.apache.kylin.engine.mr.MRUtil;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
import org.apache.kylin.job.exception.JobException;
+import org.apache.kylin.job.manager.ExecutableManager;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -56,6 +58,23 @@ public class CuboidJob extends AbstractHadoopJob {
@SuppressWarnings("rawtypes")
private Class<? extends Mapper> mapperClass;
+
+ private boolean skipped = false;
+
+ @Override
+ public boolean isSkipped() {
+ return skipped;
+ }
+
+ private boolean checkSkip(String cubingJobId) {
+ if (cubingJobId == null)
+ return false;
+
+ ExecutableManager execMgr = ExecutableManager.getInstance(KylinConfig.getInstanceFromEnv());
+ CubingJob cubingJob = (CubingJob) execMgr.getJob(cubingJobId);
+ skipped = cubingJob.isLayerCubing() == false;
+ return skipped;
+ }
@Override
public int run(String[] args) throws Exception {
@@ -72,16 +91,23 @@ public class CuboidJob extends AbstractHadoopJob {
options.addOption(OPTION_OUTPUT_PATH);
options.addOption(OPTION_NCUBOID_LEVEL);
options.addOption(OPTION_INPUT_FORMAT);
+ options.addOption(OPTION_CUBING_JOB_ID);
parseOptions(options, args);
Path output = new Path(getOptionValue(OPTION_OUTPUT_PATH));
String cubeName = getOptionValue(OPTION_CUBE_NAME).toUpperCase();
int nCuboidLevel = Integer.parseInt(getOptionValue(OPTION_NCUBOID_LEVEL));
String segmentName = getOptionValue(OPTION_SEGMENT_NAME);
+ String cubingJobId = getOptionValue(OPTION_CUBING_JOB_ID);
KylinConfig config = KylinConfig.getInstanceFromEnv();
CubeManager cubeMgr = CubeManager.getInstance(config);
CubeInstance cube = cubeMgr.getCube(cubeName);
+
+ if (checkSkip(cubingJobId)) {
+ logger.info("Skip job " + getOptionValue(OPTION_JOB_NAME) + " for " + segmentName);
+ return 0;
+ }
job = Job.getInstance(getConf(), getOptionValue(OPTION_JOB_NAME));
logger.info("Starting: " + job.getJobName());
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
index 568dd77..5fa37fa 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
@@ -19,7 +19,6 @@
package org.apache.kylin.engine.mr.steps;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
@@ -30,10 +29,8 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.hll.HyperLogLogPlusCounter;
@@ -41,8 +38,6 @@ import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
-import org.apache.kylin.cube.cuboid.Cuboid;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.engine.mr.KylinReducer;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
@@ -66,7 +61,7 @@ public class FactDistinctColumnsReducer extends KylinReducer<LongWritable, Text,
protected long baseCuboidId;
protected CubeDesc cubeDesc;
private long totalRowsBeforeMerge = 0;
- private int SAMPING_PERCENTAGE = 5;
+ private int samplingPercentage = 100;
@Override
protected void setup(Context context) throws IOException {
@@ -85,7 +80,7 @@ public class FactDistinctColumnsReducer extends KylinReducer<LongWritable, Text,
if (collectStatistics) {
baseCuboidRowCountInMappers = Lists.newArrayList();
cuboidHLLMap = Maps.newHashMap();
- SAMPING_PERCENTAGE = Integer.parseInt(context.getConfiguration().get(BatchConstants.CFG_STATISTICS_SAMPLING_PERCENT, "5"));
+ samplingPercentage = Integer.parseInt(context.getConfiguration().get(BatchConstants.CFG_STATISTICS_SAMPLING_PERCENT, "100"));
}
}
@@ -144,8 +139,15 @@ public class FactDistinctColumnsReducer extends KylinReducer<LongWritable, Text,
//output the hll info;
if (collectStatistics) {
+ long grandTotal = 0;
+ for (HyperLogLogPlusCounter hll : cuboidHLLMap.values()) {
+ grandTotal += hll.getCountEstimate();
+ }
+ double mapperOverlapRatio = (double) totalRowsBeforeMerge / grandTotal;
+
writeMapperAndCuboidStatistics(context); // for human check
- CuboidStatsUtil.writeCuboidStatistics(context.getConfiguration(), new Path(statisticsOutput), cuboidHLLMap, SAMPING_PERCENTAGE); // for CreateHTableJob
+ CuboidStatsUtil.writeCuboidStatistics(context.getConfiguration(), new Path(statisticsOutput), //
+ cuboidHLLMap, samplingPercentage, mapperOverlapRatio); // for CreateHTableJob
}
}
@@ -163,7 +165,7 @@ public class FactDistinctColumnsReducer extends KylinReducer<LongWritable, Text,
msg = "Total cuboid number: \t" + allCuboids.size();
writeLine(out, msg);
- msg = "Samping percentage: \t" + SAMPING_PERCENTAGE;
+ msg = "Samping percentage: \t" + samplingPercentage;
writeLine(out, msg);
writeLine(out, "The following statistics are collected based sampling data.");
@@ -188,7 +190,7 @@ public class FactDistinctColumnsReducer extends KylinReducer<LongWritable, Text,
writeLine(out, msg);
if (grantTotal > 0) {
- msg = "The compaction factor is: \t" + totalRowsBeforeMerge / grantTotal;
+ msg = "The mapper overlap ratio is: \t" + totalRowsBeforeMerge / grantTotal;
writeLine(out, msg);
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidJob.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidJob.java
index 8f2b810..6fec03c 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidJob.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidJob.java
@@ -18,19 +18,29 @@
package org.apache.kylin.engine.mr.steps;
+import java.io.IOException;
+import java.util.Map;
+
import org.apache.commons.cli.Options;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.ToolRunner;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.engine.mr.ByteArrayWritable;
+import org.apache.kylin.engine.mr.CubingJob;
+import org.apache.kylin.engine.mr.HadoopUtil;
import org.apache.kylin.engine.mr.IMRInput.IMRTableInputFormat;
-import org.apache.kylin.engine.mr.IMROutput2.IMRStorageOutputFormat;
import org.apache.kylin.engine.mr.MRUtil;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
+import org.apache.kylin.engine.mr.common.CubeStatsReader;
+import org.apache.kylin.job.manager.ExecutableManager;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,24 +51,49 @@ public class InMemCuboidJob extends AbstractHadoopJob {
protected static final Logger logger = LoggerFactory.getLogger(InMemCuboidJob.class);
+ private boolean skipped = false;
+
+ @Override
+ public boolean isSkipped() {
+ return skipped;
+ }
+
+ private boolean checkSkip(String cubingJobId) {
+ if (cubingJobId == null)
+ return false;
+
+ ExecutableManager execMgr = ExecutableManager.getInstance(KylinConfig.getInstanceFromEnv());
+ CubingJob cubingJob = (CubingJob) execMgr.getJob(cubingJobId);
+ skipped = cubingJob.isInMemCubing() == false;
+ return skipped;
+ }
+
@Override
public int run(String[] args) throws Exception {
Options options = new Options();
try {
options.addOption(OPTION_JOB_NAME);
- options.addOption(OPTION_JOB_FLOW_ID);
options.addOption(OPTION_CUBE_NAME);
options.addOption(OPTION_SEGMENT_NAME);
+ options.addOption(OPTION_OUTPUT_PATH);
+ options.addOption(OPTION_CUBING_JOB_ID);
parseOptions(options, args);
String cubeName = getOptionValue(OPTION_CUBE_NAME).toUpperCase();
String segmentName = getOptionValue(OPTION_SEGMENT_NAME);
+ String output = getOptionValue(OPTION_OUTPUT_PATH);
KylinConfig config = KylinConfig.getInstanceFromEnv();
CubeManager cubeMgr = CubeManager.getInstance(config);
CubeInstance cube = cubeMgr.getCube(cubeName);
CubeSegment cubeSeg = cube.getSegment(segmentName, SegmentStatusEnum.NEW);
+ String cubingJobId = getOptionValue(OPTION_CUBING_JOB_ID);
+
+ if (checkSkip(cubingJobId)) {
+ logger.info("Skip job " + getOptionValue(OPTION_JOB_NAME) + " for " + cubeSeg.getName());
+ return 0;
+ }
job = Job.getInstance(getConf(), getOptionValue(OPTION_JOB_NAME));
logger.info("Starting: " + job.getJobName());
@@ -84,8 +119,18 @@ public class InMemCuboidJob extends AbstractHadoopJob {
job.setMapOutputValueClass(ByteArrayWritable.class);
// set output
- IMRStorageOutputFormat storageOutputFormat = MRUtil.getBatchCubingOutputSide2(cubeSeg).getStorageOutputFormat();
- storageOutputFormat.configureOutput(InMemCuboidReducer.class, getOptionValue(OPTION_JOB_FLOW_ID), job);
+ job.setReducerClass(InMemCuboidReducer.class);
+ job.setNumReduceTasks(calculateReducerNum(cubeSeg));
+
+ // the cuboid file and KV class must be compatible with 0.7 version for smooth upgrade
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ job.setOutputKeyClass(Text.class);
+ job.setOutputValueClass(Text.class);
+
+ Path outputPath = new Path(output);
+ FileOutputFormat.setOutputPath(job, outputPath);
+
+ HadoopUtil.deletePath(job.getConfiguration(), outputPath);
return waitForCompletion(job);
} catch (Exception e) {
@@ -98,6 +143,31 @@ public class InMemCuboidJob extends AbstractHadoopJob {
}
}
+ private int calculateReducerNum(CubeSegment cubeSeg) throws IOException {
+ KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
+
+ Map<Long, Double> cubeSizeMap = new CubeStatsReader(cubeSeg, kylinConfig).getCuboidSizeMap();
+ double totalSizeInM = 0;
+ for (Double cuboidSize : cubeSizeMap.values()) {
+ totalSizeInM += cuboidSize;
+ }
+
+ double perReduceInputMB = kylinConfig.getDefaultHadoopJobReducerInputMB();
+
+ // number of reduce tasks
+ int numReduceTasks = (int) Math.round(totalSizeInM / perReduceInputMB);
+
+ // at least 1 reducer
+ numReduceTasks = Math.max(1, numReduceTasks);
+ // no more than 5000 reducer by default
+ numReduceTasks = Math.min(kylinConfig.getHadoopJobMaxReducerNumber(), numReduceTasks);
+
+ logger.info("Having total map input MB " + Math.round(totalSizeInM));
+ logger.info("Having per reduce MB " + perReduceInputMB);
+ logger.info("Setting " + "mapred.reduce.tasks" + "=" + numReduceTasks);
+ return numReduceTasks;
+ }
+
public static void main(String[] args) throws Exception {
InMemCuboidJob job = new InMemCuboidJob();
int exitCode = ToolRunner.run(job, args);
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidReducer.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidReducer.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidReducer.java
index db254f6..79f8b23 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidReducer.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidReducer.java
@@ -1,17 +1,18 @@
package org.apache.kylin.engine.mr.steps;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.List;
+import org.apache.hadoop.io.Text;
import org.apache.kylin.common.KylinConfig;
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.ByteArrayWritable;
-import org.apache.kylin.engine.mr.IMROutput2.IMRStorageOutputFormat;
import org.apache.kylin.engine.mr.KylinReducer;
-import org.apache.kylin.engine.mr.MRUtil;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
import org.apache.kylin.metadata.measure.MeasureAggregators;
@@ -27,13 +28,16 @@ public class InMemCuboidReducer extends KylinReducer<ByteArrayWritable, ByteArra
private static final Logger logger = LoggerFactory.getLogger(InMemCuboidReducer.class);
- private IMRStorageOutputFormat storageOutputFormat;
private MeasureCodec codec;
private MeasureAggregators aggs;
private int counter;
private Object[] input;
private Object[] result;
+
+ private Text outputKey;
+ private Text outputValue;
+ private ByteBuffer valueBuf;
@Override
protected void setup(Context context) throws IOException {
@@ -47,16 +51,16 @@ public class InMemCuboidReducer extends KylinReducer<ByteArrayWritable, ByteArra
CubeInstance cube = CubeManager.getInstance(config).getCube(cubeName);
CubeDesc cubeDesc = cube.getDescriptor();
CubeSegment cubeSeg = cube.getSegment(segmentName, SegmentStatusEnum.NEW);
- if (isMerge)
- storageOutputFormat = MRUtil.getBatchMergeOutputSide2(cubeSeg).getStorageOutputFormat();
- else
- storageOutputFormat = MRUtil.getBatchCubingOutputSide2(cubeSeg).getStorageOutputFormat();
List<MeasureDesc> measuresDescs = cubeDesc.getMeasures();
codec = new MeasureCodec(measuresDescs);
aggs = new MeasureAggregators(measuresDescs);
input = new Object[measuresDescs.size()];
result = new Object[measuresDescs.size()];
+
+ outputKey = new Text();
+ outputValue = new Text();
+ valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
}
@Override
@@ -70,8 +74,16 @@ public class InMemCuboidReducer extends KylinReducer<ByteArrayWritable, ByteArra
}
aggs.collectStates(result);
- storageOutputFormat.doReducerOutput(key, result, context);
+ // output key
+ outputKey.set(key.array(), key.offset(), key.length());
+ // output value
+ valueBuf.clear();
+ codec.encode(result, valueBuf);
+ outputValue.set(valueBuf.array(), 0, valueBuf.position());
+
+ context.write(outputKey, outputValue);
+
counter++;
if (counter % BatchConstants.COUNTER_MAX == 0) {
logger.info("Handled " + counter + " records!");
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageJob.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageJob.java
deleted file mode 100644
index 4485d17..0000000
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageJob.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
-*/
-
-package org.apache.kylin.engine.mr.steps;
-
-import org.apache.commons.cli.Options;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.cube.CubeInstance;
-import org.apache.kylin.cube.CubeManager;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.engine.mr.ByteArrayWritable;
-import org.apache.kylin.engine.mr.IMROutput2.IMRStorageInputFormat;
-import org.apache.kylin.engine.mr.IMROutput2.IMRStorageOutputFormat;
-import org.apache.kylin.engine.mr.MRUtil;
-import org.apache.kylin.engine.mr.common.BatchConstants;
-import org.apache.kylin.metadata.model.SegmentStatusEnum;
-
-/**
- */
-public class MergeCuboidFromStorageJob extends CuboidJob {
-
- @Override
- public int run(String[] args) throws Exception {
- Options options = new Options();
-
- try {
- options.addOption(OPTION_JOB_NAME);
- options.addOption(OPTION_JOB_FLOW_ID);
- options.addOption(OPTION_CUBE_NAME);
- options.addOption(OPTION_SEGMENT_NAME);
- parseOptions(options, args);
-
- String cubeName = getOptionValue(OPTION_CUBE_NAME).toUpperCase();
- String segmentName = getOptionValue(OPTION_SEGMENT_NAME).toUpperCase();
- KylinConfig config = KylinConfig.getInstanceFromEnv();
- CubeManager cubeMgr = CubeManager.getInstance(config);
- CubeInstance cube = cubeMgr.getCube(cubeName);
- CubeSegment cubeSeg = cube.getSegment(segmentName, SegmentStatusEnum.NEW);
-
- Configuration conf = this.getConf();
-
- // start job
- String jobName = getOptionValue(OPTION_JOB_NAME);
- System.out.println("Starting: " + jobName);
- job = Job.getInstance(conf, jobName);
-
- setJobClasspath(job);
-
- // add metadata to distributed cache
- attachKylinPropsAndMetadata(cube, job.getConfiguration());
-
- // set job configuration
- job.getConfiguration().set(BatchConstants.CFG_CUBE_NAME, cubeName);
- job.getConfiguration().set(BatchConstants.CFG_CUBE_SEGMENT_NAME, segmentName);
- job.getConfiguration().set(BatchConstants.CFG_IS_MERGE, "true");
-
- // configure mapper input
- IMRStorageInputFormat storageInputFormat = MRUtil.getBatchMergeInputSide2(cubeSeg).getStorageInputFormat();
- storageInputFormat.configureInput(MergeCuboidFromStorageMapper.class, ByteArrayWritable.class, ByteArrayWritable.class, job);
-
- // configure reducer output
- IMRStorageOutputFormat storageOutputFormat = MRUtil.getBatchMergeOutputSide2(cubeSeg).getStorageOutputFormat();
- storageOutputFormat.configureOutput(InMemCuboidReducer.class, getOptionValue(OPTION_JOB_FLOW_ID), job);
-
- return waitForCompletion(job);
- } catch (Exception e) {
- logger.error("error in MergeCuboidFromHBaseJob", e);
- printUsage(options);
- throw e;
- } finally {
- if (job != null)
- cleanupTempConfFile(job.getConfiguration());
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java
deleted file mode 100644
index fc616fa..0000000
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidFromStorageMapper.java
+++ /dev/null
@@ -1,261 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
-*/
-
-package org.apache.kylin.engine.mr.steps;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.List;
-
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.topn.Counter;
-import org.apache.kylin.common.topn.TopNCounter;
-import org.apache.kylin.common.util.ByteArray;
-import org.apache.kylin.common.util.BytesUtil;
-import org.apache.kylin.common.util.Pair;
-import org.apache.kylin.common.util.SplittedBytes;
-import org.apache.kylin.cube.CubeInstance;
-import org.apache.kylin.cube.CubeManager;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.common.RowKeySplitter;
-import org.apache.kylin.cube.cuboid.Cuboid;
-import org.apache.kylin.cube.kv.RowConstants;
-import org.apache.kylin.cube.kv.RowKeyEncoder;
-import org.apache.kylin.cube.kv.RowKeyEncoderProvider;
-import org.apache.kylin.cube.model.CubeDesc;
-import org.apache.kylin.dict.Dictionary;
-import org.apache.kylin.dict.DictionaryManager;
-import org.apache.kylin.engine.mr.ByteArrayWritable;
-import org.apache.kylin.engine.mr.IMROutput2.IMRStorageInputFormat;
-import org.apache.kylin.engine.mr.KylinMapper;
-import org.apache.kylin.engine.mr.MRUtil;
-import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
-import org.apache.kylin.engine.mr.common.BatchConstants;
-import org.apache.kylin.metadata.measure.MeasureCodec;
-import org.apache.kylin.metadata.model.MeasureDesc;
-import org.apache.kylin.metadata.model.SegmentStatusEnum;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-
-/**
- * @author shaoshi
- */
-public class MergeCuboidFromStorageMapper extends KylinMapper<Object, Object, ByteArrayWritable, ByteArrayWritable> {
-
- private static final Logger logger = LoggerFactory.getLogger(MergeCuboidFromStorageMapper.class);
-
- private KylinConfig config;
- private String cubeName;
- private String segmentName;
- private CubeManager cubeManager;
- private CubeInstance cube;
- private CubeDesc cubeDesc;
- private CubeSegment mergedCubeSegment;
- private CubeSegment sourceCubeSegment; // Must be unique during a mapper's life cycle
- private IMRStorageInputFormat storageInputFormat;
-
- private ByteArrayWritable outputKey = new ByteArrayWritable();
- private byte[] newKeyBodyBuf;
- private ByteArray newKeyBuf;
- private RowKeySplitter rowKeySplitter;
- private RowKeyEncoderProvider rowKeyEncoderProvider;
-
- private HashMap<TblColRef, Boolean> dimensionsNeedDict = new HashMap<TblColRef, Boolean>();
-
- private List<MeasureDesc> measureDescs;
- private ByteBuffer valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
- private MeasureCodec codec;
- private ByteArrayWritable outputValue = new ByteArrayWritable();
-
- private List<Integer> topNMeasureIdx;
-
- @Override
- protected void setup(Context context) throws IOException, InterruptedException {
- super.bindCurrentConfiguration(context.getConfiguration());
- config = AbstractHadoopJob.loadKylinPropsAndMetadata();
-
- cubeName = context.getConfiguration().get(BatchConstants.CFG_CUBE_NAME).toUpperCase();
- segmentName = context.getConfiguration().get(BatchConstants.CFG_CUBE_SEGMENT_NAME).toUpperCase();
-
- cubeManager = CubeManager.getInstance(config);
- cube = cubeManager.getCube(cubeName);
- cubeDesc = cube.getDescriptor();
- mergedCubeSegment = cube.getSegment(segmentName, SegmentStatusEnum.NEW);
- storageInputFormat = MRUtil.getBatchMergeInputSide2(mergedCubeSegment).getStorageInputFormat();
-
- newKeyBodyBuf = new byte[RowConstants.ROWKEY_BUFFER_SIZE]; // size will auto-grow
- newKeyBuf = ByteArray.allocate(RowConstants.ROWKEY_BUFFER_SIZE);
-
- sourceCubeSegment = storageInputFormat.findSourceSegment(context);
- logger.info("Source cube segment: " + sourceCubeSegment);
-
- rowKeySplitter = new RowKeySplitter(sourceCubeSegment, 65, 255);
- rowKeyEncoderProvider = new RowKeyEncoderProvider(mergedCubeSegment);
-
- measureDescs = cubeDesc.getMeasures();
- codec = new MeasureCodec(measureDescs);
-
- topNMeasureIdx = Lists.newArrayList();
- for (int i = 0; i < measureDescs.size(); i++) {
- if (measureDescs.get(i).getFunction().isTopN()) {
- topNMeasureIdx.add(i);
- }
- }
- }
-
- @Override
- public void map(Object inKey, Object inValue, Context context) throws IOException, InterruptedException {
- Pair<ByteArrayWritable, Object[]> pair = storageInputFormat.parseMapperInput(inKey, inValue);
- ByteArrayWritable key = pair.getFirst();
- Object[] value = pair.getSecond();
-
- Preconditions.checkState(key.offset() == 0);
-
- long cuboidID = rowKeySplitter.split(key.array());
- Cuboid cuboid = Cuboid.findById(cubeDesc, cuboidID);
- RowKeyEncoder rowkeyEncoder = rowKeyEncoderProvider.getRowkeyEncoder(cuboid);
-
- SplittedBytes[] splittedByteses = rowKeySplitter.getSplitBuffers();
- int bufOffset = 0;
- int bodySplitOffset = rowKeySplitter.getBodySplitOffset();
-
- for (int i = 0; i < cuboid.getColumns().size(); ++i) {
- int useSplit = i + bodySplitOffset;
- TblColRef col = cuboid.getColumns().get(i);
-
- if (this.checkNeedMerging(col)) {
- // if dictionary on fact table column, needs rewrite
- DictionaryManager dictMgr = DictionaryManager.getInstance(config);
- Dictionary<?> sourceDict = dictMgr.getDictionary(sourceCubeSegment.getDictResPath(col));
- Dictionary<?> mergedDict = dictMgr.getDictionary(mergedCubeSegment.getDictResPath(col));
-
- while (sourceDict.getSizeOfValue() > newKeyBodyBuf.length - bufOffset || //
- mergedDict.getSizeOfValue() > newKeyBodyBuf.length - bufOffset || //
- mergedDict.getSizeOfId() > newKeyBodyBuf.length - bufOffset) {
- //also use this buf to hold value before translating
- byte[] oldBuf = newKeyBodyBuf;
- newKeyBodyBuf = new byte[2 * newKeyBodyBuf.length];
- System.arraycopy(oldBuf, 0, newKeyBodyBuf, 0, oldBuf.length);
- }
-
- int idInSourceDict = BytesUtil.readUnsigned(splittedByteses[useSplit].value, 0, splittedByteses[useSplit].length);
- int size = sourceDict.getValueBytesFromId(idInSourceDict, newKeyBodyBuf, bufOffset);
-
- int idInMergedDict;
- if (size < 0) {
- idInMergedDict = mergedDict.nullId();
- } else {
- idInMergedDict = mergedDict.getIdFromValueBytes(newKeyBodyBuf, bufOffset, size);
- }
- BytesUtil.writeUnsigned(idInMergedDict, newKeyBodyBuf, bufOffset, mergedDict.getSizeOfId());
-
- bufOffset += mergedDict.getSizeOfId();
- } else {
- // keep as it is
- while (splittedByteses[useSplit].length > newKeyBodyBuf.length - bufOffset) {
- byte[] oldBuf = newKeyBodyBuf;
- newKeyBodyBuf = new byte[2 * newKeyBodyBuf.length];
- System.arraycopy(oldBuf, 0, newKeyBodyBuf, 0, oldBuf.length);
- }
-
- System.arraycopy(splittedByteses[useSplit].value, 0, newKeyBodyBuf, bufOffset, splittedByteses[useSplit].length);
- bufOffset += splittedByteses[useSplit].length;
- }
- }
-
- int fullKeySize = rowkeyEncoder.getBytesLength();
- while (newKeyBuf.array().length < fullKeySize) {
- newKeyBuf.set(new byte[newKeyBuf.length() * 2]);
- }
- newKeyBuf.set(0, fullKeySize);
-
- rowkeyEncoder.encode(new ByteArray(newKeyBodyBuf, 0, bufOffset), newKeyBuf);
- outputKey.set(newKeyBuf.array(), 0, fullKeySize);
-
- // re-encode measures if dictionary is used
- reEncodeMeasure(value);
-
- valueBuf.clear();
- codec.encode(value, valueBuf);
- outputValue.set(valueBuf.array(), 0, valueBuf.position());
-
- context.write(outputKey, outputValue);
- }
-
- private Boolean checkNeedMerging(TblColRef col) throws IOException {
- Boolean ret = dimensionsNeedDict.get(col);
- if (ret != null)
- return ret;
- else {
- ret = cubeDesc.getRowkey().isUseDictionary(col);
- if (ret) {
- String dictTable = DictionaryManager.getInstance(config).decideSourceData(cubeDesc.getModel(), cubeDesc.getRowkey().getDictionary(col), col).getTable();
- ret = cubeDesc.getFactTable().equalsIgnoreCase(dictTable);
- }
- dimensionsNeedDict.put(col, ret);
- return ret;
- }
- }
-
- @SuppressWarnings("unchecked")
- private void reEncodeMeasure(Object[] measureObjs) throws IOException, InterruptedException {
- // currently only topN uses dictionary in measure obj
- if (topNMeasureIdx.isEmpty())
- return;
-
- int bufOffset = 0;
- for (int idx : topNMeasureIdx) {
- TopNCounter<ByteArray> topNCounters = (TopNCounter<ByteArray>) measureObjs[idx];
-
- MeasureDesc measureDesc = measureDescs.get(idx);
- TblColRef colRef = measureDesc.getFunction().getTopNLiteralColumn();
- DictionaryManager dictMgr = DictionaryManager.getInstance(config);
- Dictionary<?> sourceDict = dictMgr.getDictionary(sourceCubeSegment.getDictResPath(colRef));
- Dictionary<?> mergedDict = dictMgr.getDictionary(mergedCubeSegment.getDictResPath(colRef));
-
- int topNSize = topNCounters.size();
- while (sourceDict.getSizeOfValue() * topNSize > newKeyBodyBuf.length - bufOffset || //
- mergedDict.getSizeOfValue() * topNSize > newKeyBodyBuf.length - bufOffset || //
- mergedDict.getSizeOfId() * topNSize > newKeyBodyBuf.length - bufOffset) {
- byte[] oldBuf = newKeyBodyBuf;
- newKeyBodyBuf = new byte[2 * newKeyBodyBuf.length];
- System.arraycopy(oldBuf, 0, newKeyBodyBuf, 0, oldBuf.length);
- }
-
- for (Counter<ByteArray> c : topNCounters) {
- int idInSourceDict = BytesUtil.readUnsigned(c.getItem().array(), c.getItem().offset(), c.getItem().length());
- int idInMergedDict;
- int size = sourceDict.getValueBytesFromId(idInSourceDict, newKeyBodyBuf, bufOffset);
- if (size < 0) {
- idInMergedDict = mergedDict.nullId();
- } else {
- idInMergedDict = mergedDict.getIdFromValueBytes(newKeyBodyBuf, bufOffset, size);
- }
-
- BytesUtil.writeUnsigned(idInMergedDict, newKeyBodyBuf, bufOffset, mergedDict.getSizeOfId());
- c.getItem().set(newKeyBodyBuf, bufOffset, mergedDict.getSizeOfId());
- bufOffset += mergedDict.getSizeOfId();
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeStatisticsStep.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeStatisticsStep.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeStatisticsStep.java
index 3502271..05b56aa 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeStatisticsStep.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeStatisticsStep.java
@@ -105,7 +105,7 @@ public class MergeStatisticsStep extends AbstractExecutable {
if (key.get() == 0l) {
// sampling percentage;
averageSamplingPercentage += Bytes.toInt(value.getBytes());
- } else {
+ } else if (key.get() > 0) {
HyperLogLogPlusCounter hll = new HyperLogLogPlusCounter(14);
ByteArray byteArray = new ByteArray(value.getBytes());
hll.readRegisters(byteArray.asBuffer());
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SaveStatisticsStep.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SaveStatisticsStep.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SaveStatisticsStep.java
index 2fe4dcf..b12063f 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SaveStatisticsStep.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SaveStatisticsStep.java
@@ -19,6 +19,7 @@
package org.apache.kylin.engine.mr.steps;
import java.io.IOException;
+import java.util.Random;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
@@ -29,8 +30,11 @@ import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.engine.mr.CubingJob;
+import org.apache.kylin.engine.mr.CubingJob.AlgorithmEnum;
import org.apache.kylin.engine.mr.HadoopUtil;
import org.apache.kylin.engine.mr.common.BatchConstants;
+import org.apache.kylin.engine.mr.common.CubeStatsReader;
import org.apache.kylin.job.exception.ExecuteException;
import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.ExecutableContext;
@@ -45,6 +49,7 @@ public class SaveStatisticsStep extends AbstractExecutable {
private static final String CUBE_NAME = "cubeName";
private static final String SEGMENT_ID = "segmentId";
private static final String STATISTICS_PATH = "statisticsPath";
+ private static final String CUBING_JOB_ID = "cubingJobId";
public SaveStatisticsStep() {
super();
@@ -74,6 +79,8 @@ public class SaveStatisticsStep extends AbstractExecutable {
fs.delete(statisticsFilePath, true);
}
+ decideCubingAlgorithm(newSegment, kylinConf);
+
return new ExecuteResult(ExecuteResult.State.SUCCEED, "succeed");
} catch (IOException e) {
logger.error("fail to save cuboid statistics", e);
@@ -81,6 +88,28 @@ public class SaveStatisticsStep extends AbstractExecutable {
}
}
+ private void decideCubingAlgorithm(CubeSegment seg, KylinConfig kylinConf) throws IOException {
+ String algPref = kylinConf.getCubeAlgorithm();
+ AlgorithmEnum alg;
+ if (AlgorithmEnum.INMEM.name().equalsIgnoreCase(algPref)) {
+ alg = AlgorithmEnum.INMEM;
+ } else if (AlgorithmEnum.LAYER.name().equalsIgnoreCase(algPref)) {
+ alg = AlgorithmEnum.LAYER;
+ } else if ("random".equalsIgnoreCase(algPref)) { // for testing
+ alg = new Random().nextBoolean() ? AlgorithmEnum.INMEM : AlgorithmEnum.LAYER;
+ } else { // the default
+ double threshold = kylinConf.getCubeAlgorithmAutoThreshold();
+ double mapperOverlapRatio = new CubeStatsReader(seg, kylinConf).getMapperOverlapRatioOfFirstBuild();
+ logger.info("mapperOverlapRatio for " + seg.getName() + " is " + mapperOverlapRatio + " and threshold is " + threshold);
+ alg = mapperOverlapRatio < threshold ? AlgorithmEnum.INMEM : AlgorithmEnum.LAYER;
+ }
+
+ logger.info("The cube algorithm for " + seg.getName() + " is " + alg);
+
+ CubingJob cubingJob = (CubingJob) executableManager.getJob(getCubingJobId());
+ cubingJob.setAlgorithm(alg);
+ }
+
public void setCubeName(String cubeName) {
this.setParam(CUBE_NAME, cubeName);
}
@@ -105,4 +134,12 @@ public class SaveStatisticsStep extends AbstractExecutable {
return getParam(STATISTICS_PATH);
}
+ public void setCubingJobId(String id) {
+ setParam(CUBING_JOB_ID, id);
+ }
+
+ private String getCubingJobId() {
+ return getParam(CUBING_JOB_ID);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/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 9a2610c..efd3173 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
@@ -17,12 +17,27 @@
*/
package org.apache.kylin.engine.spark;
-import com.google.common.base.*;
-import com.google.common.collect.*;
-import com.google.common.hash.HashFunction;
-import com.google.common.hash.Hasher;
-import com.google.common.hash.Hashing;
-import com.google.common.primitives.UnsignedBytes;
+import java.io.File;
+import java.io.FileFilter;
+import java.io.IOException;
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import javax.annotation.Nullable;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.OptionBuilder;
@@ -41,8 +56,10 @@ import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.ToolRunner;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.hll.HyperLogLogPlusCounter;
+import org.apache.kylin.common.util.AbstractApplication;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.ClassUtil;
+import org.apache.kylin.common.util.OptionsHelper;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
@@ -52,17 +69,19 @@ import org.apache.kylin.cube.cuboid.CuboidScheduler;
import org.apache.kylin.cube.inmemcubing.AbstractInMemCubeBuilder;
import org.apache.kylin.cube.inmemcubing.DoggedCubeBuilder;
import org.apache.kylin.cube.kv.RowConstants;
-import org.apache.kylin.cube.model.*;
+import org.apache.kylin.cube.model.CubeDesc;
+import org.apache.kylin.cube.model.CubeJoinedFlatTableDesc;
+import org.apache.kylin.cube.model.DimensionDesc;
+import org.apache.kylin.cube.model.HBaseColumnFamilyDesc;
+import org.apache.kylin.cube.model.RowKeyDesc;
import org.apache.kylin.cube.util.CubingUtils;
-import org.apache.kylin.dict.*;
import org.apache.kylin.dict.Dictionary;
import org.apache.kylin.dict.DictionaryGenerator;
import org.apache.kylin.dict.IterableDictionaryValueEnumerator;
+import org.apache.kylin.engine.mr.common.CubeStatsReader;
import org.apache.kylin.engine.spark.cube.BufferedCuboidWriter;
import org.apache.kylin.engine.spark.cube.DefaultTupleConverter;
import org.apache.kylin.engine.spark.util.IteratorUtils;
-import org.apache.kylin.common.util.AbstractApplication;
-import org.apache.kylin.common.util.OptionsHelper;
import org.apache.kylin.metadata.measure.MeasureAggregators;
import org.apache.kylin.metadata.measure.MeasureCodec;
import org.apache.kylin.metadata.model.MeasureDesc;
@@ -78,23 +97,28 @@ import org.apache.spark.SparkFiles;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
-import org.apache.spark.api.java.function.*;
+import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
+import org.apache.spark.api.java.function.Function2;
+import org.apache.spark.api.java.function.PairFlatMapFunction;
+import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.hive.HiveContext;
import org.reflections.Reflections;
+
import scala.Tuple2;
-import javax.annotation.Nullable;
-import java.io.File;
-import java.io.FileFilter;
-import java.io.IOException;
-import java.io.Serializable;
-import java.nio.ByteBuffer;
-import java.util.*;
-import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.common.hash.HashFunction;
+import com.google.common.hash.Hasher;
+import com.google.common.hash.Hashing;
+import com.google.common.primitives.UnsignedBytes;
/**
*/
@@ -453,7 +477,8 @@ public class SparkCubing extends AbstractApplication {
final KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
final CubeInstance cubeInstance = CubeManager.getInstance(kylinConfig).getCube(cubeName);
final CubeSegment cubeSegment = cubeInstance.getSegmentById(segmentId);
- final Map<Long, Long> cubeSizeMap = CreateHTableJob.getCubeRowCountMapFromCuboidStatistics(samplingResult, 100);
+ 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);
CubeHTableUtil.createHTable(cubeSegment, splitKeys);
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/monitor/StreamingMonitor.java
----------------------------------------------------------------------
diff --git a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/monitor/StreamingMonitor.java b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/monitor/StreamingMonitor.java
index a6b8a9f..9609442 100644
--- a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/monitor/StreamingMonitor.java
+++ b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/monitor/StreamingMonitor.java
@@ -92,7 +92,7 @@ public class StreamingMonitor {
private static List<CubeSegment> getSortedReadySegments(String cubeName) {
final CubeInstance cube = CubeManager.getInstance(KylinConfig.getInstanceFromEnv()).reloadCubeLocal(cubeName);
Preconditions.checkNotNull(cube);
- final List<CubeSegment> segments = cube.getSegment(SegmentStatusEnum.READY);
+ final List<CubeSegment> segments = cube.getSegments(SegmentStatusEnum.READY);
logger.info("totally " + segments.size() + " cubeSegments");
Collections.sort(segments);
return segments;
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/server/src/main/java/org/apache/kylin/rest/security/CrossDomainFilter.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/security/CrossDomainFilter.java b/server/src/main/java/org/apache/kylin/rest/security/CrossDomainFilter.java
index 79dfe50..c9167b9 100644
--- a/server/src/main/java/org/apache/kylin/rest/security/CrossDomainFilter.java
+++ b/server/src/main/java/org/apache/kylin/rest/security/CrossDomainFilter.java
@@ -53,7 +53,7 @@ public class CrossDomainFilter implements Filter {
*/
@Override
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
- if (Boolean.parseBoolean(KylinConfig.getInstanceFromEnv().getProperty("crossdomain.enable", "true"))) {
+ if (Boolean.parseBoolean(KylinConfig.getInstanceFromEnv().getOptional("crossdomain.enable", "true"))) {
((HttpServletResponse) response).addHeader("Access-Control-Allow-Origin", "*");
((HttpServletResponse) response).addHeader("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS");
((HttpServletResponse) response).addHeader("Access-Control-Allow-Headers", "Origin, No-Cache, X-Requested-With, If-Modified-Since, Pragma, Last-Modified, Cache-Control, Expires, Content-Type, X-E4M-With, Accept, Authorization");
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/server/src/main/java/org/apache/kylin/rest/service/CubeService.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/service/CubeService.java b/server/src/main/java/org/apache/kylin/rest/service/CubeService.java
index 037dace..ac4aa6b 100644
--- a/server/src/main/java/org/apache/kylin/rest/service/CubeService.java
+++ b/server/src/main/java/org/apache/kylin/rest/service/CubeService.java
@@ -585,7 +585,7 @@ public class CubeService extends BasicService {
if (desc.getRetentionRange() > 0) {
synchronized (CubeService.class) {
cube = getCubeManager().getCube(cubeName);
- List<CubeSegment> readySegs = cube.getSegment(SegmentStatusEnum.READY);
+ List<CubeSegment> readySegs = cube.getSegments(SegmentStatusEnum.READY);
long currentRange = 0;
int position = readySegs.size() - 1;
while (position >= 0) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/server/src/main/java/org/apache/kylin/rest/service/JobService.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/service/JobService.java b/server/src/main/java/org/apache/kylin/rest/service/JobService.java
index 1386656..14ae26f 100644
--- a/server/src/main/java/org/apache/kylin/rest/service/JobService.java
+++ b/server/src/main/java/org/apache/kylin/rest/service/JobService.java
@@ -143,7 +143,7 @@ public class JobService extends BasicService {
CubeSegment newSeg = getCubeManager().mergeSegments(cube, startDate, endDate, forceMergeEmptySeg);
job = EngineFactory.createBatchMergeJob(newSeg, submitter);
} else if (buildType == CubeBuildTypeEnum.REFRESH) {
- List<CubeSegment> readySegs = cube.getSegment(SegmentStatusEnum.READY);
+ List<CubeSegment> readySegs = cube.getSegments(SegmentStatusEnum.READY);
boolean segExists = false;
for (CubeSegment aSeg : readySegs) {
if (aSeg.getDateRangeStart() == startDate && aSeg.getDateRangeEnd() == endDate) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/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 dfa5e90..16a6461 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
@@ -18,26 +18,19 @@
package org.apache.kylin.storage.hbase.steps;
-import java.io.File;
-import java.io.FileOutputStream;
import java.io.IOException;
-import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import javax.annotation.Nullable;
-
import org.apache.commons.cli.Options;
import org.apache.commons.math3.primes.Primes;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
@@ -45,31 +38,22 @@ import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.ToolRunner;
import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.hll.HyperLogLogPlusCounter;
-import org.apache.kylin.common.persistence.ResourceStore;
-import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.common.util.BytesUtil;
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.cuboid.Cuboid;
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.DataType;
-import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
-import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.storage.hbase.HBaseConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Function;
-import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -110,7 +94,7 @@ public class CreateHTableJob extends AbstractHadoopJob {
try {
byte[][] splitKeys;
if (statsEnabled) {
- final Map<Long, Long> cuboidSizeMap = getCubeRowCountMapFromCuboidStatistics(cubeSegment, kylinConfig, conf);
+ final Map<Long, Double> cuboidSizeMap = new CubeStatsReader(cubeSegment, kylinConfig).getCuboidSizeMap();
splitKeys = getSplitsFromCuboidStatistics(cuboidSizeMap, kylinConfig, cubeSegment);
} else {
splitKeys = getSplits(conf, partitionFilePath);
@@ -160,50 +144,6 @@ public class CreateHTableJob extends AbstractHadoopJob {
return retValue.length == 0 ? null : retValue;
}
- public static Map<Long, Long> getCubeRowCountMapFromCuboidStatistics(CubeSegment cubeSegment, KylinConfig kylinConfig, Configuration conf) throws IOException {
- ResourceStore rs = ResourceStore.getStore(kylinConfig);
- String fileKey = cubeSegment.getStatisticsResourcePath();
- InputStream is = rs.getResource(fileKey).inputStream;
- File tempFile = null;
- FileOutputStream tempFileStream = null;
- try {
- tempFile = File.createTempFile(cubeSegment.getUuid(), ".seq");
- tempFileStream = new FileOutputStream(tempFile);
- org.apache.commons.io.IOUtils.copy(is, tempFileStream);
- } finally {
- IOUtils.closeStream(is);
- IOUtils.closeStream(tempFileStream);
- }
- Map<Long, HyperLogLogPlusCounter> counterMap = Maps.newHashMap();
-
- FileSystem fs = HadoopUtil.getFileSystem("file:///" + tempFile.getAbsolutePath());
- int samplingPercentage = 25;
- SequenceFile.Reader reader = null;
- try {
- reader = new SequenceFile.Reader(fs, new Path(tempFile.getAbsolutePath()), conf);
- LongWritable key = (LongWritable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
- BytesWritable value = (BytesWritable) ReflectionUtils.newInstance(reader.getValueClass(), conf);
- while (reader.next(key, value)) {
- if (key.get() == 0L) {
- samplingPercentage = Bytes.toInt(value.getBytes());
- } else {
- HyperLogLogPlusCounter hll = new HyperLogLogPlusCounter(14);
- ByteArray byteArray = new ByteArray(value.getBytes());
- hll.readRegisters(byteArray.asBuffer());
- counterMap.put(key.get(), hll);
- }
-
- }
- } catch (Exception e) {
- e.printStackTrace();
- throw e;
- } finally {
- IOUtils.closeStream(reader);
- tempFile.delete();
- }
- return getCubeRowCountMapFromCuboidStatistics(counterMap, samplingPercentage);
- }
-
//one region for one shard
private static byte[][] getSplitsByRegionCount(int regionCount) {
byte[][] result = new byte[regionCount - 1][];
@@ -215,51 +155,24 @@ public class CreateHTableJob extends AbstractHadoopJob {
return result;
}
- public static Map<Long, Long> getCubeRowCountMapFromCuboidStatistics(Map<Long, HyperLogLogPlusCounter> counterMap, final int samplingPercentage) throws IOException {
- Preconditions.checkArgument(samplingPercentage > 0);
- return Maps.transformValues(counterMap, new Function<HyperLogLogPlusCounter, Long>() {
- @Nullable
- @Override
- public Long apply(HyperLogLogPlusCounter input) {
- return input.getCountEstimate() * 100 / samplingPercentage;
- }
- });
- }
-
- public static byte[][] getSplitsFromCuboidStatistics(final Map<Long, Long> cubeRowCountMap, KylinConfig kylinConfig, CubeSegment cubeSegment) throws IOException {
+ public static byte[][] getSplitsFromCuboidStatistics(final Map<Long, Double> cubeSizeMap, KylinConfig kylinConfig, CubeSegment cubeSegment) throws IOException {
final CubeDesc cubeDesc = cubeSegment.getCubeDesc();
- final List<Integer> rowkeyColumnSize = Lists.newArrayList();
- final long baseCuboidId = Cuboid.getBaseCuboidId(cubeDesc);
- Cuboid baseCuboid = Cuboid.findById(cubeDesc, baseCuboidId);
- List<TblColRef> columnList = baseCuboid.getColumns();
-
- for (int i = 0; i < columnList.size(); i++) {
- logger.info("Rowkey column " + i + " length " + cubeSegment.getColumnLength(columnList.get(i)));
- rowkeyColumnSize.add(cubeSegment.getColumnLength(columnList.get(i)));
- }
-
DataModelDesc.RealizationCapacity cubeCapacity = cubeDesc.getModel().getCapacity();
int cut = kylinConfig.getHBaseRegionCut(cubeCapacity.toString());
logger.info("Cube capacity " + cubeCapacity.toString() + ", chosen cut for HTable is " + cut + "GB");
double totalSizeInM = 0;
-
- List<Long> allCuboids = Lists.newArrayList();
- allCuboids.addAll(cubeRowCountMap.keySet());
- Collections.sort(allCuboids);
-
- Map<Long, Double> cubeSizeMap = Maps.newHashMap();
- for (Map.Entry<Long, Long> entry : cubeRowCountMap.entrySet()) {
- cubeSizeMap.put(entry.getKey(), estimateCuboidStorageSize(cubeSegment, entry.getKey(), entry.getValue(), baseCuboidId, rowkeyColumnSize));
- }
-
for (Double cuboidSize : cubeSizeMap.values()) {
totalSizeInM += cuboidSize;
}
+ List<Long> allCuboids = Lists.newArrayList();
+ allCuboids.addAll(cubeSizeMap.keySet());
+ Collections.sort(allCuboids);
+
int nRegion = Math.round((float) (totalSizeInM / (cut * 1024L)));
nRegion = Math.max(kylinConfig.getHBaseRegionCountMin(), nRegion);
nRegion = Math.min(kylinConfig.getHBaseRegionCountMax(), nRegion);
@@ -350,48 +263,6 @@ public class CreateHTableJob extends AbstractHadoopJob {
}
}
- /**
- * Estimate the cuboid's size
- *
- * @return the cuboid size in M bytes
- */
- private static double estimateCuboidStorageSize(CubeSegment cubeSegment, long cuboidId, long rowCount, long baseCuboidId, List<Integer> rowKeyColumnLength) {
-
- int bytesLength = cubeSegment.getRowKeyPreambleSize();
-
- long mask = Long.highestOneBit(baseCuboidId);
- long parentCuboidIdActualLength = Long.SIZE - Long.numberOfLeadingZeros(baseCuboidId);
- for (int i = 0; i < parentCuboidIdActualLength; i++) {
- if ((mask & cuboidId) > 0) {
- bytesLength += rowKeyColumnLength.get(i); //colIO.getColumnLength(columnList.get(i));
- }
- mask = mask >> 1;
- }
-
- // add the measure length
- int space = 0;
- boolean isMemoryHungry = false;
- for (MeasureDesc measureDesc : cubeSegment.getCubeDesc().getMeasures()) {
- DataType returnType = measureDesc.getFunction().getReturnDataType();
- if (returnType.isHLLC()) {
- isMemoryHungry = true;
- }
- space += returnType.getStorageBytesEstimate();
- }
- bytesLength += space;
-
- double ret = 1.0 * bytesLength * rowCount / (1024L * 1024L);
- if (isMemoryHungry) {
- logger.info("Cube is memory hungry, storage size estimation multiply 0.05");
- ret *= 0.05;
- } else {
- logger.info("Cube is not memory hungry, storage size estimation multiply 0.25");
- ret *= 0.25;
- }
- logger.info("Cuboid " + cuboidId + " has " + rowCount + " rows, each row size is " + bytesLength + " bytes." + " Total size is " + ret + "M.");
- return ret;
- }
-
public static void main(String[] args) throws Exception {
int exitCode = ToolRunner.run(new CreateHTableJob(), args);
System.exit(exitCode);
[3/4] kylin git commit: KYLIN-1245 Put layer cubing and in-mem cubing
side by side, switch based on 'mapper overlap ratio'
Posted by li...@apache.org.
KYLIN-1245 Put layer cubing and in-mem cubing side by side, switch based on 'mapper overlap ratio'
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/2aedddf4
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/2aedddf4
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/2aedddf4
Branch: refs/heads/2.0-rc
Commit: 2aedddf4894cd09bf039bf9479a856cd38f6cf1d
Parents: 8936df1
Author: Yang Li <li...@apache.org>
Authored: Mon Dec 21 19:52:31 2015 +0800
Committer: Li, Yang <ya...@ebay.com>
Committed: Fri Dec 25 17:20:34 2015 +0800
----------------------------------------------------------------------
build/conf/kylin_hive_conf.xml | 2 +-
.../org/apache/kylin/common/KylinConfig.java | 578 +------------------
.../apache/kylin/common/KylinConfigBase.java | 506 ++++++++++++++++
.../org/apache/kylin/common/KylinConfigExt.java | 23 +
.../apache/kylin/common/util/MailService.java | 2 +-
.../kylin/common/util/MailServiceTest.java | 4 +-
.../org/apache/kylin/cube/CubeInstance.java | 10 -
.../java/org/apache/kylin/cube/CubeManager.java | 6 +-
.../org/apache/kylin/cube/model/CubeDesc.java | 3 +
.../org/apache/kylin/cube/CubeSegmentsTest.java | 2 +-
.../kylin/job/execution/AbstractExecutable.java | 4 +
.../kylin/engine/mr/BatchCubingJobBuilder2.java | 94 ++-
.../kylin/engine/mr/BatchMergeJobBuilder2.java | 35 +-
.../org/apache/kylin/engine/mr/CubingJob.java | 23 +-
.../org/apache/kylin/engine/mr/IMROutput2.java | 104 ++--
.../java/org/apache/kylin/engine/mr/MRUtil.java | 6 -
.../engine/mr/common/AbstractHadoopJob.java | 7 +-
.../kylin/engine/mr/common/CubeStatsReader.java | 246 ++++++++
.../kylin/engine/mr/common/CuboidStatsUtil.java | 15 +-
.../engine/mr/common/MapReduceExecutable.java | 6 +-
.../kylin/engine/mr/steps/BaseCuboidJob.java | 3 +-
.../apache/kylin/engine/mr/steps/CuboidJob.java | 26 +
.../mr/steps/FactDistinctColumnsReducer.java | 22 +-
.../kylin/engine/mr/steps/InMemCuboidJob.java | 78 ++-
.../engine/mr/steps/InMemCuboidReducer.java | 28 +-
.../mr/steps/MergeCuboidFromStorageJob.java | 94 ---
.../mr/steps/MergeCuboidFromStorageMapper.java | 261 ---------
.../engine/mr/steps/MergeStatisticsStep.java | 2 +-
.../engine/mr/steps/SaveStatisticsStep.java | 37 ++
.../apache/kylin/engine/spark/SparkCubing.java | 67 ++-
.../streaming/monitor/StreamingMonitor.java | 2 +-
.../kylin/rest/security/CrossDomainFilter.java | 2 +-
.../apache/kylin/rest/service/CubeService.java | 2 +-
.../apache/kylin/rest/service/JobService.java | 2 +-
.../storage/hbase/steps/CreateHTableJob.java | 143 +----
.../storage/hbase/steps/HBaseMROutput2.java | 290 ----------
.../hbase/steps/HBaseMROutput2Transition.java | 329 +----------
37 files changed, 1241 insertions(+), 1823 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/build/conf/kylin_hive_conf.xml
----------------------------------------------------------------------
diff --git a/build/conf/kylin_hive_conf.xml b/build/conf/kylin_hive_conf.xml
index 3fe17b6..9bf74d4 100644
--- a/build/conf/kylin_hive_conf.xml
+++ b/build/conf/kylin_hive_conf.xml
@@ -22,7 +22,7 @@ limitations under the License. See accompanying LICENSE file.
<property>
<name>dfs.block.size</name>
- <value>10485760</value>
+ <value>32000000</value>
<description>Want more mappers for in-mem cubing, thus smaller the DFS block size</description>
</property>
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/core-common/src/main/java/org/apache/kylin/common/KylinConfig.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/KylinConfig.java b/core-common/src/main/java/org/apache/kylin/common/KylinConfig.java
index c65ade4..bc18989 100644
--- a/core-common/src/main/java/org/apache/kylin/common/KylinConfig.java
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfig.java
@@ -27,129 +27,29 @@ import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintWriter;
-import java.io.Serializable;
import java.io.StringWriter;
import java.util.Enumeration;
import java.util.Map;
import java.util.Properties;
-import java.util.SortedSet;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.kylin.common.restclient.RestClient;
-import org.apache.kylin.common.util.CliCommandExecutor;
import org.apache.kylin.common.util.Log4jConfigurer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.Sets;
-
/**
*/
@SuppressWarnings("serial")
-public class KylinConfig implements Serializable {
-
- /*
- * NOTE: These key constants should be private or even better be removed.
- * All external access should go through public methods.
- */
- public static final String KYLIN_OWNER = "kylin.owner";
-
- public static final String KYLIN_STORAGE_URL = "kylin.storage.url";
-
- public static final String KYLIN_METADATA_URL = "kylin.metadata.url";
-
- public static final String KYLIN_REST_SERVERS = "kylin.rest.servers";
-
- public static final String KYLIN_REST_TIMEZONE = "kylin.rest.timezone";
-
- public static final String KYLIN_JOB_CONCURRENT_MAX_LIMIT = "kylin.job.concurrent.max.limit";
-
- public static final String KYLIN_JOB_YARN_APP_REST_CHECK_URL = "kylin.job.yarn.app.rest.check.status.url";
+public class KylinConfig extends KylinConfigBase {
- public static final String KYLIN_JOB_YARN_APP_REST_CHECK_INTERVAL_SECONDS = "kylin.job.yarn.app.rest.check.interval.seconds";
-
- public static final String HIVE_TABLE_LOCATION_PREFIX = "hive.table.location.";
-
- public static final String KYLIN_JOB_REMOTE_CLI_PASSWORD = "kylin.job.remote.cli.password";
-
- public static final String KYLIN_JOB_REMOTE_CLI_USERNAME = "kylin.job.remote.cli.username";
-
- public static final String KYLIN_JOB_REMOTE_CLI_HOSTNAME = "kylin.job.remote.cli.hostname";
-
- public static final String KYLIN_JOB_REMOTE_CLI_PORT = "kylin.job.remote.cli.port";
-
- public static final String KYLIN_JOB_REMOTE_CLI_WORKING_DIR = "kylin.job.remote.cli.working.dir";
-
- public static final String KYLIN_JOB_CMD_EXTRA_ARGS = "kylin.job.cmd.extra.args";
- /**
- * Toggle to indicate whether to use hive for table flattening. Default
- * true.
- */
- public static final String KYLIN_JOB_RUN_AS_REMOTE_CMD = "kylin.job.run.as.remote.cmd";
-
- public static final String KYLIN_JOB_MAPREDUCE_DEFAULT_REDUCE_COUNT_RATIO = "kylin.job.mapreduce.default.reduce.count.ratio";
-
- public static final String KYLIN_JOB_MAPREDUCE_DEFAULT_REDUCE_INPUT_MB = "kylin.job.mapreduce.default.reduce.input.mb";
-
- public static final String KYLIN_JOB_MAPREDUCE_MAX_REDUCER_NUMBER = "kylin.job.mapreduce.max.reducer.number";
-
- public static final String KYLIN_JOB_JAR = "kylin.job.jar";
-
- public static final String KYLIN_JOB_JAR_SPARK = "kylin.job.jar.spark";
-
- public static final String COPROCESSOR_LOCAL_JAR = "kylin.coprocessor.local.jar";
-
- public static final String KYLIN_JOB_LOG_DIR = "kylin.job.log.dir";
-
- public static final String KYLIN_JOB_CUBING_IN_MEM = "kylin.job.cubing.inMem";
-
- public static final String KYLIN_JOB_CUBING_IN_MEM_SAMPLING_PERCENT = "kylin.job.cubing.inMem.sampling.percent";
-
- public static final String KYLIN_HDFS_WORKING_DIR = "kylin.hdfs.working.dir";
-
- public static final String KYLIN_HBASE_CLUSTER_FS = "kylin.hbase.cluster.fs";
-
- public static final String HIVE_DATABASE_FOR_INTERMEDIATE_TABLE = "kylin.job.hive.database.for.intermediatetable";
-
- public static final String HIVE_PASSWORD = "hive.password";
-
- public static final String HIVE_USER = "hive.user";
+ private static final Logger logger = LoggerFactory.getLogger(KylinConfig.class);
- public static final String HIVE_URL = "hive.url";
- /**
- * Kylin properties file
- */
+ /** Kylin properties file name */
public static final String KYLIN_CONF_PROPERTIES_FILE = "kylin.properties";
-
- public static final String MAIL_ENABLED = "mail.enabled";
-
- public static final String MAIL_HOST = "mail.host";
-
- public static final String MAIL_USERNAME = "mail.username";
-
- public static final String MAIL_PASSWORD = "mail.password";
-
- public static final String MAIL_SENDER = "mail.sender";
-
- public static final String KYLIN_HOME = "KYLIN_HOME";
public static final String KYLIN_CONF = "KYLIN_CONF";
- public static final String HBASE_REGION_CUT_SMALL = "kylin.hbase.region.cut.small";
- public static final String HBASE_REGION_CUT_MEDIUM = "kylin.hbase.region.cut.medium";
- public static final String HBASE_REGION_CUT_LARGE = "kylin.hbase.region.cut.large";
-
- public static final String SPARK_HOME = "kylin.spark.home";
- public static final String SPARK_MASTER = "kylin.spark.master";
-
- private static final Logger logger = LoggerFactory.getLogger(KylinConfig.class);
-
- public static final String VERSION = "${project.version}";
-
- public static final String HTABLE_DEFAULT_COMPRESSION_CODEC = "kylin.hbase.default.compression.codec";
-
// static cached instances
private static KylinConfig ENV_INSTANCE = null;
@@ -249,363 +149,6 @@ public class KylinConfig implements Serializable {
return config;
}
- // ============================================================================
-
- /**
- * Find config from environment. The Search process: 1. Check the
- * $KYLIN_CONF/kylin.properties 2. Check the $KYLIN_HOME/conf/kylin.properties
- */
- private static KylinConfig loadKylinConfig() {
- Log4jConfigurer.initLogger();
-
- InputStream is = getKylinPropertiesAsInputSteam();
- if (is == null) {
- throw new IllegalArgumentException("Failed to load kylin config");
- }
- KylinConfig config = new KylinConfig();
- config.reloadKylinConfig(is);
-
- return config;
- }
-
- private volatile Properties properties = new Properties();
-
- public CliCommandExecutor getCliCommandExecutor() throws IOException {
- CliCommandExecutor exec = new CliCommandExecutor();
- if (getRunAsRemoteCommand()) {
- exec.setRunAtRemote(getRemoteHadoopCliHostname(), getRemoteHadoopCliPort(), getRemoteHadoopCliUsername(), getRemoteHadoopCliPassword());
- }
- return exec;
- }
-
- // ============================================================================
-
- public String getStorageUrl() {
- return getOptional(KYLIN_STORAGE_URL);
- }
-
- public String getHiveUrl() {
- return getOptional(HIVE_URL, "");
- }
-
- public String getHiveUser() {
- return getOptional(HIVE_USER, "");
- }
-
- public String getHivePassword() {
- return getOptional(HIVE_PASSWORD, "");
- }
-
- public String getHdfsWorkingDirectory() {
- String root = getRequired(KYLIN_HDFS_WORKING_DIR);
- if (!root.endsWith("/")) {
- root += "/";
- }
- return root + getMetadataUrlPrefix() + "/";
- }
-
- public String getHBaseClusterFs() {
- return getOptional(KYLIN_HBASE_CLUSTER_FS, "");
- }
-
- public String getKylinJobLogDir() {
- return getOptional(KYLIN_JOB_LOG_DIR, "/tmp/kylin/logs");
- }
-
- public String getKylinJobJarPath() {
- final String jobJar = getOptional(KYLIN_JOB_JAR);
- if (StringUtils.isNotEmpty(jobJar)) {
- return jobJar;
- }
- String kylinHome = getKylinHome();
- if (StringUtils.isEmpty(kylinHome)) {
- return "";
- }
- return getFileName(kylinHome + File.separator + "lib", JOB_JAR_NAME_PATTERN);
- }
-
- public String getKylinJobMRLibDir() {
- return getOptional("kylin.job.mr.lib.dir", "");
- }
-
- public String getKylinSparkJobJarPath() {
- final String jobJar = getOptional(KYLIN_JOB_JAR_SPARK);
- if (StringUtils.isNotEmpty(jobJar)) {
- return jobJar;
- }
- String kylinHome = getKylinHome();
- if (StringUtils.isEmpty(kylinHome)) {
- return "";
- }
- return getFileName(kylinHome + File.separator + "lib", SPARK_JOB_JAR_NAME_PATTERN);
- }
-
- public void overrideMRJobJarPath(String path) {
- logger.info("override " + KYLIN_JOB_JAR + " to " + path);
- System.setProperty(KYLIN_JOB_JAR, path);
- }
-
- public void overrideSparkJobJarPath(String path) {
- logger.info("override " + KYLIN_JOB_JAR_SPARK + " to " + path);
- System.setProperty(KYLIN_JOB_JAR_SPARK, path);
- }
-
- private static final Pattern COPROCESSOR_JAR_NAME_PATTERN = Pattern.compile("kylin-coprocessor-(.+)\\.jar");
- private static final Pattern JOB_JAR_NAME_PATTERN = Pattern.compile("kylin-job-(.+)\\.jar");
- private static final Pattern SPARK_JOB_JAR_NAME_PATTERN = Pattern.compile("kylin-engine-spark-(.+)\\.jar");
-
- public String getCoprocessorLocalJar() {
- final String coprocessorJar = getOptional(COPROCESSOR_LOCAL_JAR);
- if (StringUtils.isNotEmpty(coprocessorJar)) {
- return coprocessorJar;
- }
- String kylinHome = getKylinHome();
- if (StringUtils.isEmpty(kylinHome)) {
- throw new RuntimeException("getCoprocessorLocalJar needs KYLIN_HOME");
- }
- return getFileName(kylinHome + File.separator + "lib", COPROCESSOR_JAR_NAME_PATTERN);
- }
-
- private static String getFileName(String homePath, Pattern pattern) {
- File home = new File(homePath);
- SortedSet<String> files = Sets.newTreeSet();
- if (home.exists() && home.isDirectory()) {
- for (File file : home.listFiles()) {
- final Matcher matcher = pattern.matcher(file.getName());
- if (matcher.matches()) {
- files.add(file.getAbsolutePath());
- }
- }
- }
- if (files.isEmpty()) {
- throw new RuntimeException("cannot find " + pattern.toString() + " in " + homePath);
- } else {
- return files.last();
- }
- }
-
- public void overrideCoprocessorLocalJar(String path) {
- logger.info("override " + COPROCESSOR_LOCAL_JAR + " to " + path);
- System.setProperty(COPROCESSOR_LOCAL_JAR, path);
- }
-
- public double getDefaultHadoopJobReducerInputMB() {
- return Double.parseDouble(getOptional(KYLIN_JOB_MAPREDUCE_DEFAULT_REDUCE_INPUT_MB, "500"));
- }
-
- public double getDefaultHadoopJobReducerCountRatio() {
- return Double.parseDouble(getOptional(KYLIN_JOB_MAPREDUCE_DEFAULT_REDUCE_COUNT_RATIO, "1.0"));
- }
-
- public int getHadoopJobMaxReducerNumber() {
- return Integer.parseInt(getOptional(KYLIN_JOB_MAPREDUCE_MAX_REDUCER_NUMBER, "500"));
- }
-
- public boolean getRunAsRemoteCommand() {
- return Boolean.parseBoolean(getOptional(KYLIN_JOB_RUN_AS_REMOTE_CMD));
- }
-
- public int getRemoteHadoopCliPort() {
- return Integer.parseInt(getOptional(KYLIN_JOB_REMOTE_CLI_PORT, "22"));
- }
-
- public String getRemoteHadoopCliHostname() {
- return getOptional(KYLIN_JOB_REMOTE_CLI_HOSTNAME);
- }
-
- public String getRemoteHadoopCliUsername() {
- return getOptional(KYLIN_JOB_REMOTE_CLI_USERNAME);
- }
-
- public String getRemoteHadoopCliPassword() {
- return getOptional(KYLIN_JOB_REMOTE_CLI_PASSWORD);
- }
-
- public String getCliWorkingDir() {
- return getOptional(KYLIN_JOB_REMOTE_CLI_WORKING_DIR);
- }
-
- public String getMapReduceCmdExtraArgs() {
- return getOptional(KYLIN_JOB_CMD_EXTRA_ARGS);
- }
-
- public String getOverrideHiveTableLocation(String table) {
- return getOptional(HIVE_TABLE_LOCATION_PREFIX + table.toUpperCase());
- }
-
- public String getYarnStatusCheckUrl() {
- return getOptional(KYLIN_JOB_YARN_APP_REST_CHECK_URL, null);
- }
-
- public int getYarnStatusCheckIntervalSeconds() {
- return Integer.parseInt(getOptional(KYLIN_JOB_YARN_APP_REST_CHECK_INTERVAL_SECONDS, "60"));
- }
-
- public int getMaxConcurrentJobLimit() {
- return Integer.parseInt(getOptional(KYLIN_JOB_CONCURRENT_MAX_LIMIT, "10"));
- }
-
- public String getTimeZone() {
- return getOptional(KYLIN_REST_TIMEZONE, "PST");
- }
-
- public String[] getRestServers() {
- return getOptionalStringArray(KYLIN_REST_SERVERS);
- }
-
- public String getAdminDls() {
- return getOptional("kylin.job.admin.dls", null);
- }
-
- public long getJobStepTimeout() {
- return Long.parseLong(getOptional("kylin.job.step.timeout", String.valueOf(2 * 60 * 60)));
- }
-
- public String getServerMode() {
- return this.getOptional("kylin.server.mode", "all");
- }
-
- public int getDictionaryMaxCardinality() {
- return Integer.parseInt(getOptional("kylin.dictionary.max.cardinality", "5000000"));
- }
-
- public int getTableSnapshotMaxMB() {
- return Integer.parseInt(getOptional("kylin.table.snapshot.max_mb", "300"));
- }
-
- public int getHBaseRegionCountMin() {
- return Integer.parseInt(getOptional("kylin.hbase.region.count.min", "1"));
- }
-
- public int getHBaseRegionCountMax() {
- return Integer.parseInt(getOptional("kylin.hbase.region.count.max", "500"));
- }
-
- public int getScanThreshold() {
- return Integer.parseInt(getOptional("kylin.query.scan.threshold", "10000000"));
- }
-
- public boolean getQueryRunLocalCoprocessor() {
- return Boolean.parseBoolean(getOptional("kylin.query.run.local.coprocessor", "false"));
- }
-
- public Long getQueryDurationCacheThreshold() {
- return Long.parseLong(this.getOptional("kylin.query.cache.threshold.duration", String.valueOf(2000)));
- }
-
- public Long getQueryScanCountCacheThreshold() {
- return Long.parseLong(this.getOptional("kylin.query.cache.threshold.scancount", String.valueOf(10 * 1024)));
- }
-
- public long getQueryMemBudget() {
- return Long.parseLong(this.getOptional("kylin.query.mem.budget", String.valueOf(3L * 1024 * 1024 * 1024)));
- }
-
- public boolean isQuerySecureEnabled() {
- return Boolean.parseBoolean(this.getOptional("kylin.query.security.enabled", "false"));
- }
-
- public boolean isQueryCacheEnabled() {
- return Boolean.parseBoolean(this.getOptional("kylin.query.cache.enabled", "true"));
- }
-
- public int getHBaseKeyValueSize() {
- return Integer.parseInt(this.getOptional("kylin.hbase.client.keyvalue.maxsize", "10485760"));
- }
-
- public int getHBaseScanCacheRows() {
- return Integer.parseInt(this.getOptional("kylin.hbase.scan.cache_rows", "1024"));
- }
-
- public int getHBaseScanMaxResultSize() {
- return Integer.parseInt(this.getOptional("kylin.hbase.scan.max_result_size", "" + (5 * 1024 * 1024))); // 5 MB
- }
-
- public boolean isCubingInMem() {
- return Boolean.parseBoolean(this.getOptional(KYLIN_JOB_CUBING_IN_MEM, "false"));
- }
-
- public int getCubingInMemSamplingPercent() {
- int percent = Integer.parseInt(this.getOptional(KYLIN_JOB_CUBING_IN_MEM_SAMPLING_PERCENT, "100"));
- percent = Math.max(percent, 1);
- percent = Math.min(percent, 100);
- return percent;
- }
-
- public String getHbaseDefaultCompressionCodec() {
- return getOptional(HTABLE_DEFAULT_COMPRESSION_CODEC, "");
- }
-
- public boolean isHiveKeepFlatTable() {
- return Boolean.parseBoolean(this.getOptional("kylin.hive.keep.flat.table", "false"));
- }
-
- private String getOptional(String prop) {
- final String property = System.getProperty(prop);
- return property != null ? property : properties.getProperty(prop);
- }
-
- private String[] getOptionalStringArray(String prop) {
- final String property = getOptional(prop);
- if (!StringUtils.isBlank(property)) {
- return property.split("\\s*,\\s*");
- } else {
- return new String[] {};
- }
- }
-
- private String getOptional(String prop, String dft) {
- final String property = System.getProperty(prop);
- return property != null ? property : properties.getProperty(prop, dft);
- }
-
- private String getRequired(String prop) {
- final String property = System.getProperty(prop);
- if (property != null) {
- return property;
- }
- String r = properties.getProperty(prop);
- if (StringUtils.isEmpty(r)) {
- throw new IllegalArgumentException("missing '" + prop + "' in conf/kylin_instance.properties");
- }
- return r;
- }
-
- void reloadKylinConfig(InputStream is) {
- Properties newProperties = new Properties();
- try {
- newProperties.load(is);
- } catch (IOException e) {
- throw new RuntimeException("Cannot load kylin config.", e);
- } finally {
- IOUtils.closeQuietly(is);
- }
- this.properties = newProperties;
- }
-
- public void writeProperties(File file) throws IOException {
- FileOutputStream fos = null;
- try {
- fos = new FileOutputStream(file);
- properties.store(fos, file.getAbsolutePath());
- } finally {
- IOUtils.closeQuietly(fos);
- }
- }
-
- public static String getKylinHome() {
- String kylinHome = System.getenv(KYLIN_HOME);
- if (StringUtils.isEmpty(kylinHome)) {
- logger.warn("KYLIN_HOME was not set");
- return kylinHome;
- }
- return kylinHome;
- }
-
- public void printProperties() throws IOException {
- properties.list(System.out);
- }
-
private static File getKylinProperties() {
String kylinConfHome = System.getProperty(KYLIN_CONF);
if (!StringUtils.isEmpty(kylinConfHome)) {
@@ -679,68 +222,33 @@ public class KylinConfig implements Serializable {
return new File(path, KYLIN_CONF_PROPERTIES_FILE);
}
- public String getMetadataUrl() {
- return getOptional(KYLIN_METADATA_URL);
- }
-
- public String getMetadataUrlPrefix() {
- String hbaseMetadataUrl = getMetadataUrl();
- String defaultPrefix = "kylin_metadata";
+ /**
+ * Find config from environment. The Search process: 1. Check the
+ * $KYLIN_CONF/kylin.properties 2. Check the $KYLIN_HOME/conf/kylin.properties
+ */
+ private static KylinConfig loadKylinConfig() {
+ Log4jConfigurer.initLogger();
- if (org.apache.commons.lang3.StringUtils.containsIgnoreCase(hbaseMetadataUrl, "@hbase")) {
- int cut = hbaseMetadataUrl.indexOf('@');
- String tmp = cut < 0 ? defaultPrefix : hbaseMetadataUrl.substring(0, cut);
- return tmp;
- } else {
- return defaultPrefix;
+ InputStream is = getKylinPropertiesAsInputSteam();
+ if (is == null) {
+ throw new IllegalArgumentException("Failed to load kylin config");
}
- }
-
- public void setMetadataUrl(String metadataUrl) {
- properties.setProperty(KYLIN_METADATA_URL, metadataUrl);
- }
-
- public void setStorageUrl(String storageUrl) {
- properties.setProperty(KYLIN_STORAGE_URL, storageUrl);
- }
-
- public String getHiveDatabaseForIntermediateTable() {
- return this.getOptional(HIVE_DATABASE_FOR_INTERMEDIATE_TABLE, "default");
- }
-
- public String getKylinOwner() {
- return this.getOptional(KYLIN_OWNER, "");
- }
-
- public void setRunAsRemoteCommand(String v) {
- properties.setProperty(KYLIN_JOB_RUN_AS_REMOTE_CMD, v);
- }
-
- public void setRemoteHadoopCliHostname(String v) {
- properties.setProperty(KYLIN_JOB_REMOTE_CLI_HOSTNAME, v);
- }
-
- public void setRemoteHadoopCliUsername(String v) {
- properties.setProperty(KYLIN_JOB_REMOTE_CLI_USERNAME, v);
- }
+ KylinConfig config = new KylinConfig();
+ config.reloadKylinConfig(is);
- public void setRemoteHadoopCliPassword(String v) {
- properties.setProperty(KYLIN_JOB_REMOTE_CLI_PASSWORD, v);
+ return config;
}
- public String getProperty(String key, String defaultValue) {
- return properties.getProperty(key, defaultValue);
- }
+ // ============================================================================
- /**
- * Set a new key:value into the kylin config.
- *
- * @param key
- * @param value
- */
- public void setProperty(String key, String value) {
- logger.info("Kylin Config was updated with " + key + " : " + value);
- properties.setProperty(key, value);
+ public void writeProperties(File file) throws IOException {
+ FileOutputStream fos = null;
+ try {
+ fos = new FileOutputStream(file);
+ getAllProperties().store(fos, file.getAbsolutePath());
+ } finally {
+ IOUtils.closeQuietly(fos);
+ }
}
public String getConfigAsString() throws IOException {
@@ -750,42 +258,12 @@ public class KylinConfig implements Serializable {
}
private void list(PrintWriter out) {
- for (Enumeration e = properties.keys() ; e.hasMoreElements() ;) {
- String key = (String)e.nextElement();
- String val = (String)properties.get(key);
+ Properties props = getAllProperties();
+ for (Enumeration<?> e = props.keys(); e.hasMoreElements();) {
+ String key = (String) e.nextElement();
+ String val = (String) props.get(key);
out.println(key + "=" + val);
}
}
- public String getSparkHome() {
- return properties.getProperty(SPARK_HOME);
- }
-
- public String getSparkMaster() {
- return properties.getProperty(SPARK_MASTER);
- }
-
- public int getHBaseRegionCut(String capacity) {
- String cut;
- switch (capacity) {
- case "SMALL":
- cut = getProperty(HBASE_REGION_CUT_SMALL, "10");
- break;
- case "MEDIUM":
- cut = getProperty(HBASE_REGION_CUT_MEDIUM, "20");
- break;
- case "LARGE":
- cut = getProperty(HBASE_REGION_CUT_LARGE, "100");
- break;
- default:
- throw new IllegalArgumentException("Capacity not recognized: " + capacity);
- }
-
- return Integer.valueOf(cut);
- }
-
- public String toString() {
- return getMetadataUrl();
- }
-
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/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
new file mode 100644
index 0000000..302a2db
--- /dev/null
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
@@ -0,0 +1,506 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.kylin.common;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Serializable;
+import java.util.Properties;
+import java.util.SortedSet;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.kylin.common.util.CliCommandExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Sets;
+
+@SuppressWarnings("serial")
+/**
+ * An abstract class to encapsulate access to a set of 'properties'.
+ * Subclass can override methods in this class to extend the content of the 'properties',
+ * with some override values for example.
+ */
+public class KylinConfigBase implements Serializable {
+
+ private static final Logger logger = LoggerFactory.getLogger(KylinConfigBase.class);
+
+ /*
+ * DON'T DEFINE CONSTANTS FOR PROPERTY KEYS!
+ *
+ * For 1), no external need to access property keys, all accesses are by public methods.
+ * For 2), it's cumbersome to maintain constants at top and code at bottom.
+ * For 3), key literals usually appear only once.
+ */
+
+ public static String getKylinHome() {
+ String kylinHome = System.getenv("KYLIN_HOME");
+ if (StringUtils.isEmpty(kylinHome)) {
+ logger.warn("KYLIN_HOME was not set");
+ }
+ return kylinHome;
+ }
+
+ // ============================================================================
+
+ private volatile Properties properties = new Properties();
+
+ public String getOptional(String prop) {
+ return getOptional(prop, null);
+ }
+
+ public String getOptional(String prop, String dft) {
+ final String property = System.getProperty(prop);
+ return property != null ? property : properties.getProperty(prop, dft);
+ }
+
+ protected String[] getOptionalStringArray(String prop, String[] dft) {
+ final String property = getOptional(prop);
+ if (!StringUtils.isBlank(property)) {
+ return property.split("\\s*,\\s*");
+ } else {
+ return dft;
+ }
+ }
+
+ public String getRequired(String prop) {
+ String r = getOptional(prop);
+ if (StringUtils.isEmpty(r)) {
+ throw new IllegalArgumentException("missing '" + prop + "' in conf/kylin_instance.properties");
+ }
+ return r;
+ }
+
+ /**
+ * Use with care, properties should be read-only. This is for testing mostly.
+ */
+ public void setProperty(String key, String value) {
+ logger.info("Kylin Config was updated with " + key + " : " + value);
+ properties.setProperty(key, value);
+ }
+
+ protected Properties getAllProperties() {
+ return properties;
+ }
+
+ protected void reloadKylinConfig(InputStream is) {
+ Properties newProperties = new Properties();
+ try {
+ newProperties.load(is);
+ } catch (IOException e) {
+ throw new RuntimeException("Cannot load kylin config.", e);
+ } finally {
+ IOUtils.closeQuietly(is);
+ }
+ this.properties = newProperties;
+ }
+
+ // ============================================================================
+
+ public String getMetadataUrl() {
+ return getOptional("kylin.metadata.url");
+ }
+
+ public void setMetadataUrl(String metadataUrl) {
+ setProperty("kylin.metadata.url", metadataUrl);
+ }
+
+ public String getMetadataUrlPrefix() {
+ String hbaseMetadataUrl = getMetadataUrl();
+ String defaultPrefix = "kylin_metadata";
+
+ if (org.apache.commons.lang3.StringUtils.containsIgnoreCase(hbaseMetadataUrl, "@hbase")) {
+ int cut = hbaseMetadataUrl.indexOf('@');
+ String tmp = cut < 0 ? defaultPrefix : hbaseMetadataUrl.substring(0, cut);
+ return tmp;
+ } else {
+ return defaultPrefix;
+ }
+ }
+
+ public String getServerMode() {
+ return this.getOptional("kylin.server.mode", "all");
+ }
+
+ public String getStorageUrl() {
+ return getOptional("kylin.storage.url");
+ }
+
+ public void setStorageUrl(String storageUrl) {
+ setProperty("kylin.storage.url", storageUrl);
+ }
+
+ /** was for route to hive, not used any more */
+ @Deprecated
+ public String getHiveUrl() {
+ return getOptional("hive.url", "");
+ }
+
+ /** was for route to hive, not used any more */
+ @Deprecated
+ public String getHiveUser() {
+ return getOptional("hive.user", "");
+ }
+
+ /** was for route to hive, not used any more */
+ @Deprecated
+ public String getHivePassword() {
+ return getOptional("hive.password", "");
+ }
+
+ public String getHdfsWorkingDirectory() {
+ String root = getRequired("kylin.hdfs.working.dir");
+ if (!root.endsWith("/")) {
+ root += "/";
+ }
+ return root + getMetadataUrlPrefix() + "/";
+ }
+
+ public CliCommandExecutor getCliCommandExecutor() throws IOException {
+ CliCommandExecutor exec = new CliCommandExecutor();
+ if (getRunAsRemoteCommand()) {
+ exec.setRunAtRemote(getRemoteHadoopCliHostname(), getRemoteHadoopCliPort(), getRemoteHadoopCliUsername(), getRemoteHadoopCliPassword());
+ }
+ return exec;
+ }
+
+ public String getHBaseClusterFs() {
+ return getOptional("kylin.hbase.cluster.fs", "");
+ }
+
+ public String getKylinJobLogDir() {
+ return getOptional("kylin.job.log.dir", "/tmp/kylin/logs");
+ }
+
+ public String getKylinJobJarPath() {
+ final String jobJar = getOptional("kylin.job.jar");
+ if (StringUtils.isNotEmpty(jobJar)) {
+ return jobJar;
+ }
+ String kylinHome = getKylinHome();
+ if (StringUtils.isEmpty(kylinHome)) {
+ return "";
+ }
+ return getFileName(kylinHome + File.separator + "lib", JOB_JAR_NAME_PATTERN);
+ }
+
+ public void overrideMRJobJarPath(String path) {
+ logger.info("override " + "kylin.job.jar" + " to " + path);
+ System.setProperty("kylin.job.jar", path);
+ }
+
+ public String getKylinJobMRLibDir() {
+ return getOptional("kylin.job.mr.lib.dir", "");
+ }
+
+ public String getKylinSparkJobJarPath() {
+ final String jobJar = getOptional("kylin.job.jar.spark");
+ if (StringUtils.isNotEmpty(jobJar)) {
+ return jobJar;
+ }
+ String kylinHome = getKylinHome();
+ if (StringUtils.isEmpty(kylinHome)) {
+ return "";
+ }
+ return getFileName(kylinHome + File.separator + "lib", SPARK_JOB_JAR_NAME_PATTERN);
+ }
+
+ public void overrideSparkJobJarPath(String path) {
+ logger.info("override " + "kylin.job.jar.spark" + " to " + path);
+ System.setProperty("kylin.job.jar.spark", path);
+ }
+
+ private static final Pattern COPROCESSOR_JAR_NAME_PATTERN = Pattern.compile("kylin-coprocessor-(.+)\\.jar");
+ private static final Pattern JOB_JAR_NAME_PATTERN = Pattern.compile("kylin-job-(.+)\\.jar");
+ private static final Pattern SPARK_JOB_JAR_NAME_PATTERN = Pattern.compile("kylin-engine-spark-(.+)\\.jar");
+
+ public String getCoprocessorLocalJar() {
+ final String coprocessorJar = getOptional("kylin.coprocessor.local.jar");
+ if (StringUtils.isNotEmpty(coprocessorJar)) {
+ return coprocessorJar;
+ }
+ String kylinHome = getKylinHome();
+ if (StringUtils.isEmpty(kylinHome)) {
+ throw new RuntimeException("getCoprocessorLocalJar needs KYLIN_HOME");
+ }
+ return getFileName(kylinHome + File.separator + "lib", COPROCESSOR_JAR_NAME_PATTERN);
+ }
+
+ public void overrideCoprocessorLocalJar(String path) {
+ logger.info("override " + "kylin.coprocessor.local.jar" + " to " + path);
+ System.setProperty("kylin.coprocessor.local.jar", path);
+ }
+
+ private static String getFileName(String homePath, Pattern pattern) {
+ File home = new File(homePath);
+ SortedSet<String> files = Sets.newTreeSet();
+ if (home.exists() && home.isDirectory()) {
+ for (File file : home.listFiles()) {
+ final Matcher matcher = pattern.matcher(file.getName());
+ if (matcher.matches()) {
+ files.add(file.getAbsolutePath());
+ }
+ }
+ }
+ if (files.isEmpty()) {
+ throw new RuntimeException("cannot find " + pattern.toString() + " in " + homePath);
+ } else {
+ return files.last();
+ }
+ }
+
+ public double getDefaultHadoopJobReducerInputMB() {
+ return Double.parseDouble(getOptional("kylin.job.mapreduce.default.reduce.input.mb", "500"));
+ }
+
+ public double getDefaultHadoopJobReducerCountRatio() {
+ return Double.parseDouble(getOptional("kylin.job.mapreduce.default.reduce.count.ratio", "1.0"));
+ }
+
+ public int getHadoopJobMaxReducerNumber() {
+ return Integer.parseInt(getOptional("kylin.job.mapreduce.max.reducer.number", "500"));
+ }
+
+ public boolean getRunAsRemoteCommand() {
+ return Boolean.parseBoolean(getOptional("kylin.job.run.as.remote.cmd"));
+ }
+
+ public void setRunAsRemoteCommand(String v) {
+ setProperty("kylin.job.run.as.remote.cmd", v);
+ }
+
+ public int getRemoteHadoopCliPort() {
+ return Integer.parseInt(getOptional("kylin.job.remote.cli.port", "22"));
+ }
+
+ public String getRemoteHadoopCliHostname() {
+ return getOptional("kylin.job.remote.cli.hostname");
+ }
+
+ public void setRemoteHadoopCliHostname(String v) {
+ setProperty("kylin.job.remote.cli.hostname", v);
+ }
+
+ public String getRemoteHadoopCliUsername() {
+ return getOptional("kylin.job.remote.cli.username");
+ }
+
+ public void setRemoteHadoopCliUsername(String v) {
+ setProperty("kylin.job.remote.cli.username", v);
+ }
+
+ public String getRemoteHadoopCliPassword() {
+ return getOptional("kylin.job.remote.cli.password");
+ }
+
+ public void setRemoteHadoopCliPassword(String v) {
+ setProperty("kylin.job.remote.cli.password", v);
+ }
+
+ public String getCliWorkingDir() {
+ return getOptional("kylin.job.remote.cli.working.dir");
+ }
+
+ public String getMapReduceCmdExtraArgs() {
+ return getOptional("kylin.job.cmd.extra.args");
+ }
+
+ public String getOverrideHiveTableLocation(String table) {
+ return getOptional("hive.table.location." + table.toUpperCase());
+ }
+
+ public String getYarnStatusCheckUrl() {
+ return getOptional("kylin.job.yarn.app.rest.check.status.url", null);
+ }
+
+ public int getYarnStatusCheckIntervalSeconds() {
+ return Integer.parseInt(getOptional("kylin.job.yarn.app.rest.check.interval.seconds", "60"));
+ }
+
+ public int getMaxConcurrentJobLimit() {
+ return Integer.parseInt(getOptional("kylin.job.concurrent.max.limit", "10"));
+ }
+
+ public String getTimeZone() {
+ return getOptional("kylin.rest.timezone", "PST");
+ }
+
+ public String[] getRestServers() {
+ return getOptionalStringArray("kylin.rest.servers", new String[0]);
+ }
+
+ public String getAdminDls() {
+ return getOptional("kylin.job.admin.dls", null);
+ }
+
+ public long getJobStepTimeout() {
+ return Long.parseLong(getOptional("kylin.job.step.timeout", String.valueOf(2 * 60 * 60)));
+ }
+
+ public String getCubeAlgorithm() {
+ return getOptional("kylin.cube.algorithm", "auto");
+ }
+
+ public double getCubeAlgorithmAutoThreshold() {
+ return Double.parseDouble(getOptional("kylin.cube.algorithm.auto.threshold", "8"));
+ }
+
+ public int getDictionaryMaxCardinality() {
+ return Integer.parseInt(getOptional("kylin.dictionary.max.cardinality", "5000000"));
+ }
+
+ public int getTableSnapshotMaxMB() {
+ return Integer.parseInt(getOptional("kylin.table.snapshot.max_mb", "300"));
+ }
+
+ public int getHBaseRegionCut(String capacity) {
+ String cut;
+ switch (capacity) {
+ case "SMALL":
+ cut = getOptional("kylin.hbase.region.cut.small", "10");
+ break;
+ case "MEDIUM":
+ cut = getOptional("kylin.hbase.region.cut.medium", "20");
+ break;
+ case "LARGE":
+ cut = getOptional("kylin.hbase.region.cut.large", "100");
+ break;
+ default:
+ throw new IllegalArgumentException("Capacity not recognized: " + capacity);
+ }
+
+ return Integer.valueOf(cut);
+ }
+
+ public int getHBaseRegionCountMin() {
+ return Integer.parseInt(getOptional("kylin.hbase.region.count.min", "1"));
+ }
+
+ public int getHBaseRegionCountMax() {
+ return Integer.parseInt(getOptional("kylin.hbase.region.count.max", "500"));
+ }
+
+ public int getScanThreshold() {
+ return Integer.parseInt(getOptional("kylin.query.scan.threshold", "10000000"));
+ }
+
+ public boolean getQueryRunLocalCoprocessor() {
+ return Boolean.parseBoolean(getOptional("kylin.query.run.local.coprocessor", "false"));
+ }
+
+ public Long getQueryDurationCacheThreshold() {
+ return Long.parseLong(this.getOptional("kylin.query.cache.threshold.duration", String.valueOf(2000)));
+ }
+
+ public Long getQueryScanCountCacheThreshold() {
+ return Long.parseLong(this.getOptional("kylin.query.cache.threshold.scancount", String.valueOf(10 * 1024)));
+ }
+
+ public long getQueryMemBudget() {
+ return Long.parseLong(this.getOptional("kylin.query.mem.budget", String.valueOf(3L * 1024 * 1024 * 1024)));
+ }
+
+ public boolean isQuerySecureEnabled() {
+ return Boolean.parseBoolean(this.getOptional("kylin.query.security.enabled", "false"));
+ }
+
+ public boolean isQueryCacheEnabled() {
+ return Boolean.parseBoolean(this.getOptional("kylin.query.cache.enabled", "true"));
+ }
+
+ public boolean isQueryIgnoreUnknownFunction() {
+ return Boolean.parseBoolean(this.getOptional("kylin.query.ignore_unknown_function", "false"));
+ }
+
+ public int getHBaseKeyValueSize() {
+ return Integer.parseInt(this.getOptional("kylin.hbase.client.keyvalue.maxsize", "10485760"));
+ }
+
+ public int getHBaseScanCacheRows() {
+ return Integer.parseInt(this.getOptional("kylin.hbase.scan.cache_rows", "1024"));
+ }
+
+ public int getHBaseScanMaxResultSize() {
+ return Integer.parseInt(this.getOptional("kylin.hbase.scan.max_result_size", "" + (5 * 1024 * 1024))); // 5 MB
+ }
+
+ public int getCubingInMemSamplingPercent() {
+ int percent = Integer.parseInt(this.getOptional("kylin.job.cubing.inMem.sampling.percent", "100"));
+ percent = Math.max(percent, 1);
+ percent = Math.min(percent, 100);
+ return percent;
+ }
+
+ public String getHbaseDefaultCompressionCodec() {
+ return getOptional("kylin.hbase.default.compression.codec", "");
+ }
+
+ public boolean isHiveKeepFlatTable() {
+ return Boolean.parseBoolean(this.getOptional("kylin.hive.keep.flat.table", "false"));
+ }
+
+ public String getHiveDatabaseForIntermediateTable() {
+ return this.getOptional("kylin.job.hive.database.for.intermediatetable", "default");
+ }
+
+ public String getKylinOwner() {
+ return this.getOptional("kylin.owner", "");
+ }
+
+ public String getSparkHome() {
+ return getRequired("kylin.spark.home");
+ }
+
+ public String getSparkMaster() {
+ return getRequired("kylin.spark.master");
+ }
+
+ public boolean isMailEnabled() {
+ return Boolean.parseBoolean(getOptional("mail.enabled", "false"));
+ }
+
+ public void setMailEnabled(boolean enable) {
+ setProperty("mail.enabled", "" + enable);
+ }
+
+ public String getMailHost() {
+ return getOptional("mail.host", "");
+ }
+
+ public String getMailUsername() {
+ return getOptional("mail.username", "");
+ }
+
+ public String getMailPassword() {
+ return getOptional("mail.password", "");
+ }
+
+ public String getMailSender() {
+ return getOptional("mail.sender", "");
+ }
+
+ public String toString() {
+ return getMetadataUrl();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/core-common/src/main/java/org/apache/kylin/common/KylinConfigExt.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/KylinConfigExt.java b/core-common/src/main/java/org/apache/kylin/common/KylinConfigExt.java
new file mode 100644
index 0000000..8af35b6
--- /dev/null
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfigExt.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.kylin.common;
+
+public class KylinConfigExt {
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/core-common/src/main/java/org/apache/kylin/common/util/MailService.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/MailService.java b/core-common/src/main/java/org/apache/kylin/common/util/MailService.java
index 3b0a4e1..9a4ec64 100644
--- a/core-common/src/main/java/org/apache/kylin/common/util/MailService.java
+++ b/core-common/src/main/java/org/apache/kylin/common/util/MailService.java
@@ -42,7 +42,7 @@ public class MailService {
private static final Log logger = LogFactory.getLog(MailService.class);
public MailService(KylinConfig config) {
- this("true".equalsIgnoreCase(config.getProperty(KylinConfig.MAIL_ENABLED, "false")), config.getProperty(KylinConfig.MAIL_HOST, ""), config.getProperty(KylinConfig.MAIL_USERNAME, ""), config.getProperty(KylinConfig.MAIL_PASSWORD, ""), config.getProperty(KylinConfig.MAIL_SENDER, ""));
+ this(config.isMailEnabled(), config.getMailHost(), config.getMailUsername(), config.getMailPassword(), config.getMailSender());
}
private MailService(boolean enabled, String host, String username, String password, String sender) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/core-common/src/test/java/org/apache/kylin/common/util/MailServiceTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/util/MailServiceTest.java b/core-common/src/test/java/org/apache/kylin/common/util/MailServiceTest.java
index bc4d7cf..df06221 100644
--- a/core-common/src/test/java/org/apache/kylin/common/util/MailServiceTest.java
+++ b/core-common/src/test/java/org/apache/kylin/common/util/MailServiceTest.java
@@ -51,8 +51,8 @@ public class MailServiceTest extends LocalFileMetadataTestCase {
boolean sent = sendTestEmail(mailservice);
assert sent;
- // set mail.enabled=false, and run again, this time should be no mail delviered
- config.setProperty(KylinConfig.MAIL_ENABLED, "false");
+ // set mail.enabled=false, and run again, this time should be no mail delivered
+ config.setMailEnabled(false);
mailservice = new MailService(config);
sent = sendTestEmail(mailservice);
assert !sent;
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/core-cube/src/main/java/org/apache/kylin/cube/CubeInstance.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/CubeInstance.java b/core-cube/src/main/java/org/apache/kylin/cube/CubeInstance.java
index 2208136..41c1f41 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/CubeInstance.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/CubeInstance.java
@@ -297,16 +297,6 @@ public class CubeInstance extends RootPersistentEntity implements IRealization,
return result;
}
- public List<CubeSegment> getSegment(SegmentStatusEnum status) {
- List<CubeSegment> result = Lists.newArrayList();
- for (CubeSegment segment : segments) {
- if (segment.getStatus() == status) {
- result.add(segment);
- }
- }
- return result;
- }
-
public CubeSegment getSegment(String name, SegmentStatusEnum status) {
for (CubeSegment segment : segments) {
if ((null != segment.getName() && segment.getName().equals(name)) && segment.getStatus() == status) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java b/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
index 4592b15..c5ea962 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
@@ -270,7 +270,7 @@ public class CubeManager implements IRealizationProvider {
}
private boolean validateReadySegments(CubeInstance cube) {
- final List<CubeSegment> readySegments = cube.getSegment(SegmentStatusEnum.READY);
+ final List<CubeSegment> readySegments = cube.getSegments(SegmentStatusEnum.READY);
if (readySegments.size() == 0) {
return true;
}
@@ -474,7 +474,7 @@ public class CubeManager implements IRealizationProvider {
}
private Pair<Long, Long> alignMergeRange(CubeInstance cube, long startDate, long endDate) {
- List<CubeSegment> readySegments = cube.getSegment(SegmentStatusEnum.READY);
+ List<CubeSegment> readySegments = cube.getSegments(SegmentStatusEnum.READY);
if (readySegments.isEmpty()) {
throw new IllegalStateException("there are no segments in ready state");
}
@@ -609,7 +609,7 @@ public class CubeManager implements IRealizationProvider {
return null;
}
- List<CubeSegment> readySegments = Lists.newArrayList(cube.getSegment(SegmentStatusEnum.READY));
+ List<CubeSegment> readySegments = Lists.newArrayList(cube.getSegments(SegmentStatusEnum.READY));
if (readySegments.size() == 0) {
logger.debug("Cube " + cube.getName() + " has no ready segment to merge");
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
index e78dea5..6f3dc00 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
@@ -26,6 +26,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -124,6 +125,8 @@ public class CubeDesc extends RootPersistentEntity {
private int engineType = IEngineAware.ID_MR_V1;
@JsonProperty("storage_type")
private int storageType = IStorageAware.ID_HBASE;
+ @JsonProperty("override_kylin_properties")
+ private LinkedHashMap<String, String> overrideKylinProps = new LinkedHashMap<String, String>();
private Map<String, Map<String, TblColRef>> columnMap = new HashMap<String, Map<String, TblColRef>>();
private LinkedHashSet<TblColRef> allColumns = new LinkedHashSet<TblColRef>();
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/core-cube/src/test/java/org/apache/kylin/cube/CubeSegmentsTest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/test/java/org/apache/kylin/cube/CubeSegmentsTest.java b/core-cube/src/test/java/org/apache/kylin/cube/CubeSegmentsTest.java
index 6d1d39b..c1f55d1 100644
--- a/core-cube/src/test/java/org/apache/kylin/cube/CubeSegmentsTest.java
+++ b/core-cube/src/test/java/org/apache/kylin/cube/CubeSegmentsTest.java
@@ -65,7 +65,7 @@ public class CubeSegmentsTest extends LocalFileMetadataTestCase {
// assert one ready segment
assertEquals(1, cube.getSegments().size());
- CubeSegment seg = cube.getSegment(SegmentStatusEnum.READY).get(0);
+ CubeSegment seg = cube.getSegments(SegmentStatusEnum.READY).get(0);
assertEquals(SegmentStatusEnum.READY, seg.getStatus());
// append again, for non-partitioned cube, it becomes a full refresh
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java b/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
index 7ba328d..18d3193 100644
--- a/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
@@ -269,6 +269,10 @@ public abstract class AbstractExecutable implements Executable, Idempotent {
protected final void addExtraInfo(String key, String value) {
executableManager.addJobInfo(getId(), key, value);
}
+
+ protected final Map<String, String> getExtraInfo() {
+ return executableManager.getOutput(getId()).getExtra();
+ }
public final void setStartTime(long time) {
addExtraInfo(START_TIME, time + "");
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder2.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder2.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder2.java
index f8fbc33..86439a8 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder2.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder2.java
@@ -19,10 +19,13 @@
package org.apache.kylin.engine.mr;
import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.model.RowKeyDesc;
import org.apache.kylin.engine.mr.IMRInput.IMRBatchCubingInputSide;
import org.apache.kylin.engine.mr.IMROutput2.IMRBatchCubingOutputSide2;
import org.apache.kylin.engine.mr.common.MapReduceExecutable;
+import org.apache.kylin.engine.mr.steps.BaseCuboidJob;
import org.apache.kylin.engine.mr.steps.InMemCuboidJob;
+import org.apache.kylin.engine.mr.steps.NDCuboidJob;
import org.apache.kylin.engine.mr.steps.SaveStatisticsStep;
import org.apache.kylin.job.constant.ExecutableConstants;
import org.slf4j.Logger;
@@ -37,14 +40,15 @@ public class BatchCubingJobBuilder2 extends JobBuilderSupport {
public BatchCubingJobBuilder2(CubeSegment newSegment, String submitter) {
super(newSegment, submitter);
this.inputSide = MRUtil.getBatchCubingInputSide(seg);
- this.outputSide = MRUtil.getBatchCubingOutputSide2((CubeSegment)seg);
+ this.outputSide = MRUtil.getBatchCubingOutputSide2((CubeSegment) seg);
}
public CubingJob build() {
logger.info("MR_V2 new job to BUILD segment " + seg);
-
- final CubingJob result = CubingJob.createBuildJob((CubeSegment)seg, submitter, config);
+
+ final CubingJob result = CubingJob.createBuildJob((CubeSegment) seg, submitter, config);
final String jobId = result.getId();
+ final String cuboidRootPath = getCuboidRootPath(jobId);
// Phase 1: Create Flat Table
inputSide.addStepPhase1_CreateFlatTable(result);
@@ -56,8 +60,9 @@ public class BatchCubingJobBuilder2 extends JobBuilderSupport {
outputSide.addStepPhase2_BuildDictionary(result);
// Phase 3: Build Cube
- result.addTask(createInMemCubingStep(jobId));
- outputSide.addStepPhase3_BuildCube(result);
+ addLayerCubingSteps(result, jobId, cuboidRootPath); // layer cubing, only selected algorithm will execute
+ result.addTask(createInMemCubingStep(jobId, cuboidRootPath)); // inmem cubing, only selected algorithm will execute
+ outputSide.addStepPhase3_BuildCube(result, cuboidRootPath);
// Phase 4: Update Metadata & Cleanup
result.addTask(createUpdateCubeInfoAfterBuildStep(jobId));
@@ -67,28 +72,44 @@ public class BatchCubingJobBuilder2 extends JobBuilderSupport {
return result;
}
+ private void addLayerCubingSteps(final CubingJob result, final String jobId, final String cuboidRootPath) {
+ RowKeyDesc rowKeyDesc = ((CubeSegment) seg).getCubeDesc().getRowkey();
+ final int groupRowkeyColumnsCount = ((CubeSegment) seg).getCubeDesc().getBuildLevel();
+ final int totalRowkeyColumnsCount = rowKeyDesc.getRowKeyColumns().length;
+ final String[] cuboidOutputTempPath = getCuboidOutputPaths(cuboidRootPath, totalRowkeyColumnsCount, groupRowkeyColumnsCount);
+ // base cuboid step
+ result.addTask(createBaseCuboidStep(cuboidOutputTempPath, jobId));
+ // n dim cuboid steps
+ for (int i = 1; i <= groupRowkeyColumnsCount; i++) {
+ int dimNum = totalRowkeyColumnsCount - i;
+ result.addTask(createNDimensionCuboidStep(cuboidOutputTempPath, dimNum, totalRowkeyColumnsCount, jobId));
+ }
+ }
+
private SaveStatisticsStep createSaveStatisticsStep(String jobId) {
SaveStatisticsStep result = new SaveStatisticsStep();
result.setName(ExecutableConstants.STEP_NAME_SAVE_STATISTICS);
result.setCubeName(seg.getRealization().getName());
result.setSegmentId(seg.getUuid());
result.setStatisticsPath(getStatisticsPath(jobId));
+ result.setCubingJobId(jobId);
return result;
}
- private MapReduceExecutable createInMemCubingStep(String jobId) {
+ private MapReduceExecutable createInMemCubingStep(String jobId, String cuboidRootPath) {
// base cuboid job
MapReduceExecutable cubeStep = new MapReduceExecutable();
StringBuilder cmd = new StringBuilder();
- appendMapReduceParameters(cmd, ((CubeSegment)seg).getCubeDesc().getModel());
+ appendMapReduceParameters(cmd, ((CubeSegment) seg).getCubeDesc().getModel());
cubeStep.setName(ExecutableConstants.STEP_NAME_BUILD_IN_MEM_CUBE);
appendExecCmdParameters(cmd, "cubename", seg.getRealization().getName());
appendExecCmdParameters(cmd, "segmentname", seg.getName());
+ appendExecCmdParameters(cmd, "output", cuboidRootPath);
appendExecCmdParameters(cmd, "jobname", "Kylin_Cube_Builder_" + seg.getRealization().getName());
- appendExecCmdParameters(cmd, "jobflowid", jobId);
+ appendExecCmdParameters(cmd, "cubingJobId", jobId);
cubeStep.setMapReduceParams(cmd.toString());
cubeStep.setMapReduceJobClass(InMemCuboidJob.class);
@@ -96,4 +117,61 @@ public class BatchCubingJobBuilder2 extends JobBuilderSupport {
return cubeStep;
}
+ private MapReduceExecutable createBaseCuboidStep(String[] cuboidOutputTempPath, String jobId) {
+ // base cuboid job
+ MapReduceExecutable baseCuboidStep = new MapReduceExecutable();
+
+ StringBuilder cmd = new StringBuilder();
+ appendMapReduceParameters(cmd, ((CubeSegment) seg).getCubeDesc().getModel());
+
+ baseCuboidStep.setName(ExecutableConstants.STEP_NAME_BUILD_BASE_CUBOID);
+
+ appendExecCmdParameters(cmd, "cubename", seg.getRealization().getName());
+ appendExecCmdParameters(cmd, "segmentname", seg.getName());
+ appendExecCmdParameters(cmd, "input", "FLAT_TABLE"); // marks flat table input
+ appendExecCmdParameters(cmd, "output", cuboidOutputTempPath[0]);
+ appendExecCmdParameters(cmd, "jobname", "Kylin_Base_Cuboid_Builder_" + seg.getRealization().getName());
+ appendExecCmdParameters(cmd, "level", "0");
+ appendExecCmdParameters(cmd, "cubingJobId", jobId);
+
+ baseCuboidStep.setMapReduceParams(cmd.toString());
+ baseCuboidStep.setMapReduceJobClass(BaseCuboidJob.class);
+ baseCuboidStep.setCounterSaveAs(CubingJob.SOURCE_RECORD_COUNT + "," + CubingJob.SOURCE_SIZE_BYTES);
+ return baseCuboidStep;
+ }
+
+ private MapReduceExecutable createNDimensionCuboidStep(String[] cuboidOutputTempPath, int dimNum, int totalRowkeyColumnCount, String jobId) {
+ // ND cuboid job
+ MapReduceExecutable ndCuboidStep = new MapReduceExecutable();
+
+ ndCuboidStep.setName(ExecutableConstants.STEP_NAME_BUILD_N_D_CUBOID + " : " + dimNum + "-Dimension");
+ StringBuilder cmd = new StringBuilder();
+
+ appendMapReduceParameters(cmd, ((CubeSegment) seg).getCubeDesc().getModel());
+ appendExecCmdParameters(cmd, "cubename", seg.getRealization().getName());
+ appendExecCmdParameters(cmd, "segmentname", seg.getName());
+ appendExecCmdParameters(cmd, "input", cuboidOutputTempPath[totalRowkeyColumnCount - dimNum - 1]);
+ appendExecCmdParameters(cmd, "output", cuboidOutputTempPath[totalRowkeyColumnCount - dimNum]);
+ appendExecCmdParameters(cmd, "jobname", "Kylin_ND-Cuboid_Builder_" + seg.getRealization().getName() + "_Step");
+ appendExecCmdParameters(cmd, "level", "" + (totalRowkeyColumnCount - dimNum));
+ appendExecCmdParameters(cmd, "cubingJobId", jobId);
+
+ ndCuboidStep.setMapReduceParams(cmd.toString());
+ ndCuboidStep.setMapReduceJobClass(NDCuboidJob.class);
+ return ndCuboidStep;
+ }
+
+ private String[] getCuboidOutputPaths(String cuboidRootPath, int totalRowkeyColumnCount, int groupRowkeyColumnsCount) {
+ String[] paths = new String[groupRowkeyColumnsCount + 1];
+ for (int i = 0; i <= groupRowkeyColumnsCount; i++) {
+ int dimNum = totalRowkeyColumnCount - i;
+ if (dimNum == totalRowkeyColumnCount) {
+ paths[i] = cuboidRootPath + "base_cuboid";
+ } else {
+ paths[i] = cuboidRootPath + dimNum + "d_cuboid";
+ }
+ }
+ return paths;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder2.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder2.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder2.java
index 48a717f..008d489 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder2.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder2.java
@@ -23,7 +23,7 @@ import java.util.List;
import org.apache.kylin.common.util.StringUtil;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.engine.mr.common.MapReduceExecutable;
-import org.apache.kylin.engine.mr.steps.MergeCuboidFromStorageJob;
+import org.apache.kylin.engine.mr.steps.MergeCuboidJob;
import org.apache.kylin.engine.mr.steps.MergeStatisticsStep;
import org.apache.kylin.job.constant.ExecutableConstants;
import org.slf4j.Logger;
@@ -39,23 +39,24 @@ public class BatchMergeJobBuilder2 extends JobBuilderSupport {
public BatchMergeJobBuilder2(CubeSegment mergeSegment, String submitter) {
super(mergeSegment, submitter);
- this.outputSide = MRUtil.getBatchMergeOutputSide2((CubeSegment)seg);
+ this.outputSide = MRUtil.getBatchMergeOutputSide2((CubeSegment) seg);
}
public CubingJob build() {
logger.info("MR_V2 new job to MERGE segment " + seg);
-
- final CubeSegment cubeSegment = (CubeSegment)seg;
+
+ final CubeSegment cubeSegment = (CubeSegment) seg;
final CubingJob result = CubingJob.createMergeJob(cubeSegment, submitter, config);
final String jobId = result.getId();
+ final String cuboidRootPath = getCuboidRootPath(jobId);
final List<CubeSegment> mergingSegments = cubeSegment.getCubeInstance().getMergingSegments(cubeSegment);
Preconditions.checkState(mergingSegments.size() > 1, "there should be more than 2 segments to merge");
final List<String> mergingSegmentIds = Lists.newArrayList();
- final List<String> mergingHTables = Lists.newArrayList();
+ final List<String> mergingCuboidPaths = Lists.newArrayList();
for (CubeSegment merging : mergingSegments) {
mergingSegmentIds.add(merging.getUuid());
- mergingHTables.add(merging.getStorageLocationIdentifier());
+ mergingCuboidPaths.add(getCuboidRootPath(merging) + "*");
}
// Phase 1: Merge Dictionary
@@ -63,10 +64,10 @@ public class BatchMergeJobBuilder2 extends JobBuilderSupport {
result.addTask(createMergeStatisticsStep(cubeSegment, mergingSegmentIds, getStatisticsPath(jobId)));
outputSide.addStepPhase1_MergeDictionary(result);
- // Phase 2: Merge Cube
- String formattedTables = StringUtil.join(mergingHTables, ",");
- result.addTask(createMergeCuboidDataFromStorageStep(formattedTables, jobId));
- outputSide.addStepPhase2_BuildCube(result);
+ // Phase 2: Merge Cube Files
+ String formattedPath = StringUtil.join(mergingCuboidPaths, ",");
+ result.addTask(createMergeCuboidDataStep(cubeSegment, formattedPath, cuboidRootPath));
+ outputSide.addStepPhase2_BuildCube(result, cuboidRootPath);
// Phase 3: Update Metadata & Cleanup
result.addTask(createUpdateCubeInfoAfterMergeStep(mergingSegmentIds, jobId));
@@ -85,20 +86,20 @@ public class BatchMergeJobBuilder2 extends JobBuilderSupport {
return result;
}
- private MapReduceExecutable createMergeCuboidDataFromStorageStep(String inputTableNames, String jobId) {
+ private MapReduceExecutable createMergeCuboidDataStep(CubeSegment seg, String inputPath, String outputPath) {
MapReduceExecutable mergeCuboidDataStep = new MapReduceExecutable();
mergeCuboidDataStep.setName(ExecutableConstants.STEP_NAME_MERGE_CUBOID);
StringBuilder cmd = new StringBuilder();
- appendMapReduceParameters(cmd, ((CubeSegment)seg).getCubeDesc().getModel());
- appendExecCmdParameters(cmd, "cubename", seg.getRealization().getName());
+ appendMapReduceParameters(cmd, seg.getRealization().getDataModelDesc());
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
appendExecCmdParameters(cmd, "segmentname", seg.getName());
- appendExecCmdParameters(cmd, "jobname", "Kylin_Merge_Cuboid_" + seg.getRealization().getName() + "_Step");
- appendExecCmdParameters(cmd, "jobflowid", jobId);
+ appendExecCmdParameters(cmd, "input", inputPath);
+ appendExecCmdParameters(cmd, "output", outputPath);
+ appendExecCmdParameters(cmd, "jobname", "Kylin_Merge_Cuboid_" + seg.getCubeInstance().getName() + "_Step");
mergeCuboidDataStep.setMapReduceParams(cmd.toString());
- mergeCuboidDataStep.setMapReduceJobClass(MergeCuboidFromStorageJob.class);
- mergeCuboidDataStep.setCounterSaveAs(",," + CubingJob.CUBE_SIZE_BYTES);
+ mergeCuboidDataStep.setMapReduceJobClass(MergeCuboidJob.class);
return mergeCuboidDataStep;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/CubingJob.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/CubingJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/CubingJob.java
index bee030f..32e0f21 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/CubingJob.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/CubingJob.java
@@ -46,6 +46,10 @@ import org.apache.kylin.job.execution.Output;
/**
*/
public class CubingJob extends DefaultChainedExecutable {
+
+ public static enum AlgorithmEnum {
+ LAYER, INMEM
+ }
// KEYS of Output.extraInfo map, info passed across job steps
public static final String SOURCE_RECORD_COUNT = "sourceRecordCount";
@@ -169,7 +173,24 @@ public class CubingJob extends DefaultChainedExecutable {
public void setMapReduceWaitTime(long t) {
addExtraInfo(MAP_REDUCE_WAIT_TIME, t + "");
}
+
+ public void setAlgorithm(AlgorithmEnum alg) {
+ addExtraInfo("algorithm", alg.name());
+ }
+
+ public AlgorithmEnum getAlgorithm() {
+ String alg = getExtraInfo().get("algorithm");
+ return alg == null ? null : AlgorithmEnum.valueOf(alg);
+ }
+ public boolean isLayerCubing() {
+ return AlgorithmEnum.LAYER == getAlgorithm();
+ }
+
+ public boolean isInMemCubing() {
+ return AlgorithmEnum.INMEM == getAlgorithm();
+ }
+
public long findSourceRecordCount() {
return Long.parseLong(findExtraInfo(SOURCE_RECORD_COUNT, "0"));
}
@@ -206,5 +227,5 @@ public class CubingJob extends DefaultChainedExecutable {
}
return dft;
}
-
+
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/IMROutput2.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/IMROutput2.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/IMROutput2.java
index 3ad51c5..844eb07 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/IMROutput2.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/IMROutput2.java
@@ -1,13 +1,23 @@
-package org.apache.kylin.engine.mr;
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
-import java.io.IOException;
+package org.apache.kylin.engine.mr;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.kylin.common.util.Pair;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.job.execution.DefaultChainedExecutable;
@@ -17,93 +27,59 @@ public interface IMROutput2 {
public IMRBatchCubingOutputSide2 getBatchCubingOutputSide(CubeSegment seg);
/**
- * Participate the batch cubing flow as the output side.
+ * Participate the batch cubing flow as the output side. Responsible for saving
+ * the cuboid output to storage at the end of Phase 3.
*
* - Phase 1: Create Flat Table
* - Phase 2: Build Dictionary
- * - Phase 3: Build Cube (with StorageOutputFormat)
+ * - Phase 3: Build Cube
* - Phase 4: Update Metadata & Cleanup
*/
public interface IMRBatchCubingOutputSide2 {
- /** Return an output format for Phase 3: Build Cube MR */
- public IMRStorageOutputFormat getStorageOutputFormat();
-
/** Add step that executes after build dictionary and before build cube. */
public void addStepPhase2_BuildDictionary(DefaultChainedExecutable jobFlow);
- /** Add step that executes after build cube. */
- public void addStepPhase3_BuildCube(DefaultChainedExecutable jobFlow);
-
- /** Add step that does any necessary clean up. */
- public void addStepPhase4_Cleanup(DefaultChainedExecutable jobFlow);
- }
-
- public IMRBatchMergeInputSide2 getBatchMergeInputSide(CubeSegment seg);
-
- public interface IMRBatchMergeInputSide2 {
- public IMRStorageInputFormat getStorageInputFormat();
- }
-
- /** Read in a cube as input of merge. Configure the input file format of mapper. */
- @SuppressWarnings("rawtypes")
- public interface IMRStorageInputFormat {
-
- /** Configure MR mapper class and input file format. */
- public void configureInput(Class<? extends Mapper> mapperClz, Class<? extends WritableComparable> outputKeyClz, Class<? extends Writable> outputValueClz, Job job) throws IOException;
-
- /** Given a mapper context, figure out which segment the mapper reads from. */
- public CubeSegment findSourceSegment(Mapper.Context context) throws IOException;
-
/**
- * Read in a row of cuboid. Given the input KV, de-serialize back cuboid ID, dimensions, and measures.
+ * Add step that saves cuboids from HDFS to storage.
*
- * @return <code>ByteArrayWritable</code> is the cuboid ID (8 bytes) + dimension values in dictionary encoding
- * <code>Object[]</code> is the measure values in order of <code>CubeDesc.getMeasures()</code>
+ * The cuboid output is a directory of sequence files, where key is CUBOID+D1+D2+..+Dn,
+ * value is M1+M2+..+Mm. CUBOID is 8 bytes cuboid ID; Dx is dimension value with
+ * dictionary encoding; Mx is measure value serialization form.
*/
- public Pair<ByteArrayWritable, Object[]> parseMapperInput(Object inKey, Object inValue);
+ public void addStepPhase3_BuildCube(DefaultChainedExecutable jobFlow, String cuboidRootPath);
+
+ /** Add step that does any necessary clean up. */
+ public void addStepPhase4_Cleanup(DefaultChainedExecutable jobFlow);
}
/** Return a helper to participate in batch merge job flow. */
public IMRBatchMergeOutputSide2 getBatchMergeOutputSide(CubeSegment seg);
/**
- * Participate the batch merge flow as the output side.
+ * Participate the batch cubing flow as the output side. Responsible for saving
+ * the cuboid output to storage at the end of Phase 2.
*
* - Phase 1: Merge Dictionary
- * - Phase 2: Merge Cube (with StorageInputFormat & StorageOutputFormat)
+ * - Phase 2: Merge Cube
* - Phase 3: Update Metadata & Cleanup
*/
public interface IMRBatchMergeOutputSide2 {
- /** Return an input format for Phase 2: Merge Cube MR */
- public IMRStorageOutputFormat getStorageOutputFormat();
-
/** Add step that executes after merge dictionary and before merge cube. */
public void addStepPhase1_MergeDictionary(DefaultChainedExecutable jobFlow);
- /** Add step that executes after merge cube. */
- public void addStepPhase2_BuildCube(DefaultChainedExecutable jobFlow);
+ /**
+ * Add step that saves cuboid output from HDFS to storage.
+ *
+ * The cuboid output is a directory of sequence files, where key is CUBOID+D1+D2+..+Dn,
+ * value is M1+M2+..+Mm. CUBOID is 8 bytes cuboid ID; Dx is dimension value with
+ * dictionary encoding; Mx is measure value serialization form.
+ */
+ public void addStepPhase2_BuildCube(DefaultChainedExecutable jobFlow, String cuboidRootPath);
/** Add step that does any necessary clean up. */
public void addStepPhase3_Cleanup(DefaultChainedExecutable jobFlow);
}
- /** Write out a cube. Configure the output file format of reducer and do the actual K-V output. */
- @SuppressWarnings("rawtypes")
- public interface IMRStorageOutputFormat {
-
- /** Configure MR reducer class and output file format. */
- public void configureOutput(Class<? extends Reducer> reducer, String jobFlowId, Job job) throws IOException;
-
- /**
- * Write out a row of cuboid. Given the cuboid ID, dimensions, and measures, serialize in whatever
- * way and output to reducer context.
- *
- * @param key The cuboid ID (8 bytes) + dimension values in dictionary encoding
- * @param value The measure values in order of <code>CubeDesc.getMeasures()</code>
- * @param context The reducer context output goes to
- */
- public void doReducerOutput(ByteArrayWritable key, Object[] value, Reducer.Context context) throws IOException, InterruptedException;
- }
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/MRUtil.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/MRUtil.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/MRUtil.java
index 55fa9e2..41c8b6b 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/MRUtil.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/MRUtil.java
@@ -7,12 +7,10 @@ import org.apache.kylin.engine.mr.IMRInput.IMRTableInputFormat;
import org.apache.kylin.engine.mr.IMROutput.IMRBatchCubingOutputSide;
import org.apache.kylin.engine.mr.IMROutput.IMRBatchMergeOutputSide;
import org.apache.kylin.engine.mr.IMROutput2.IMRBatchCubingOutputSide2;
-import org.apache.kylin.engine.mr.IMROutput2.IMRBatchMergeInputSide2;
import org.apache.kylin.engine.mr.IMROutput2.IMRBatchMergeOutputSide2;
import org.apache.kylin.invertedindex.IISegment;
import org.apache.kylin.metadata.MetadataManager;
import org.apache.kylin.metadata.model.TableDesc;
-import org.apache.kylin.metadata.realization.IRealization;
import org.apache.kylin.metadata.realization.IRealizationSegment;
import org.apache.kylin.source.SourceFactory;
import org.apache.kylin.storage.StorageFactory;
@@ -47,10 +45,6 @@ public class MRUtil {
return StorageFactory.createEngineAdapter(seg, IMROutput2.class).getBatchCubingOutputSide(seg);
}
- public static IMRBatchMergeInputSide2 getBatchMergeInputSide2(CubeSegment seg) {
- return StorageFactory.createEngineAdapter(seg, IMROutput2.class).getBatchMergeInputSide(seg);
- }
-
public static IMRBatchMergeOutputSide2 getBatchMergeOutputSide2(CubeSegment seg) {
return StorageFactory.createEngineAdapter(seg, IMROutput2.class).getBatchMergeOutputSide(seg);
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2aedddf4/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/AbstractHadoopJob.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/AbstractHadoopJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/AbstractHadoopJob.java
index f031f76..748bac9 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/AbstractHadoopJob.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/AbstractHadoopJob.java
@@ -72,8 +72,8 @@ public abstract class AbstractHadoopJob extends Configured implements Tool {
protected static final Logger logger = LoggerFactory.getLogger(AbstractHadoopJob.class);
protected static final Option OPTION_JOB_NAME = OptionBuilder.withArgName("name").hasArg().isRequired(true).withDescription("Job name. For exmaple, Kylin_Cuboid_Builder-clsfd_v2_Step_22-D)").create("jobname");
- protected static final Option OPTION_JOB_FLOW_ID = OptionBuilder.withArgName("job flow ID").hasArg().isRequired(true).withDescription("job flow ID").create("jobflowid");
protected static final Option OPTION_CUBE_NAME = OptionBuilder.withArgName("name").hasArg().isRequired(true).withDescription("Cube name. For exmaple, flat_item_cube").create("cubename");
+ protected static final Option OPTION_CUBING_JOB_ID = OptionBuilder.withArgName("cubingJobId").hasArg().isRequired(false).withDescription("ID of cubing job executable").create("cubingJobId");
protected static final Option OPTION_II_NAME = OptionBuilder.withArgName("name").hasArg().isRequired(true).withDescription("II name. For exmaple, some_ii").create("iiname");
protected static final Option OPTION_SEGMENT_NAME = OptionBuilder.withArgName("name").hasArg().isRequired(true).withDescription("Cube segment name)").create("segmentname");
protected static final Option OPTION_TABLE_NAME = OptionBuilder.withArgName("name").hasArg().isRequired(true).withDescription("Hive table name.").create("tablename");
@@ -477,4 +477,9 @@ public abstract class AbstractHadoopJob extends Configured implements Tool {
return this.job;
}
+ // tells MapReduceExecutable to skip this job
+ public boolean isSkipped() {
+ return false;
+ }
+
}