You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by ka...@apache.org on 2017/05/24 10:42:35 UTC
[04/50] [abbrv] kylin git commit: KYLIN-2623 Move output(Hbase) code
to outputside
KYLIN-2623 Move output(Hbase) code to outputside
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/a38b02df
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/a38b02df
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/a38b02df
Branch: refs/heads/KYLIN-2606
Commit: a38b02df0387541684aa1689e044927af650f1c7
Parents: edc4d4c
Author: Roger Shi <ro...@hotmail.com>
Authored: Tue May 16 13:50:23 2017 +0800
Committer: liyang-gmt8 <li...@apache.org>
Committed: Tue May 16 14:03:50 2017 +0800
----------------------------------------------------------------------
.../org/apache/kylin/engine/mr/IMROutput2.java | 33 +++++-
.../engine/mr/common/AbstractHadoopJob.java | 6 +-
.../apache/kylin/engine/mr/steps/CuboidJob.java | 37 +++----
.../kylin/engine/mr/steps/InMemCuboidJob.java | 55 ++--------
.../engine/mr/steps/LayerReducerNumSizing.java | 80 --------------
.../kylin/engine/mr/steps/MergeCuboidJob.java | 37 ++++---
.../engine/mr/steps/MergeCuboidMapper.java | 35 +-----
.../kylin/engine/mr/steps/ReducerNumSizing.java | 106 +++++++++++++++++++
.../apache/kylin/source/hive/HiveMRInput.java | 2 -
.../apache/kylin/source/kafka/KafkaMRInput.java | 1 -
.../hbase/steps/HBaseMROutput2Transition.java | 97 +++++++++++++++++
11 files changed, 284 insertions(+), 205 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/a38b02df/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 603f207..69bba0a 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
@@ -18,11 +18,14 @@
package org.apache.kylin.engine.mr;
+import java.util.List;
+
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.job.execution.DefaultChainedExecutable;
-import java.util.List;
-
public interface IMROutput2 {
/** Return a helper to participate in batch cubing job flow. */
@@ -53,6 +56,19 @@ public interface IMROutput2 {
/** Add step that does any necessary clean up. */
public void addStepPhase4_Cleanup(DefaultChainedExecutable jobFlow);
+
+ public IMROutputFormat getOuputFormat();
+
+ }
+
+ public interface IMROutputFormat {
+
+ /** Configure the InputFormat of given job. */
+ public void configureJobInput(Job job, String input) throws Exception;
+
+ /** Configure the OutputFormat of given job. */
+ public void configureJobOutput(Job job, String output, CubeSegment segment, int level) throws Exception;
+
}
/** Return a helper to participate in batch merge job flow. */
@@ -82,6 +98,19 @@ public interface IMROutput2 {
/** Add step that does any necessary clean up. */
public void addStepPhase3_Cleanup(DefaultChainedExecutable jobFlow);
+
+ public IMRMergeOutputFormat getOuputFormat();
+ }
+
+ public interface IMRMergeOutputFormat {
+
+ /** Configure the InputFormat of given job. */
+ public void configureJobInput(Job job, String input) throws Exception;
+
+ /** Configure the OutputFormat of given job. */
+ public void configureJobOutput(Job job, String output, CubeSegment segment) throws Exception;
+
+ public CubeSegment findSourceSegment(FileSplit fileSplit, CubeInstance cube);
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/a38b02df/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 44686d6..764cbdd 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
@@ -559,7 +559,7 @@ public abstract class AbstractHadoopJob extends Configured implements Tool {
HadoopUtil.deletePath(conf, path);
}
- protected double getTotalMapInputMB() throws ClassNotFoundException, IOException, InterruptedException, JobException {
+ public static double getTotalMapInputMB(Job job) throws ClassNotFoundException, IOException, InterruptedException, JobException {
if (job == null) {
throw new JobException("Job is null");
}
@@ -576,6 +576,10 @@ public abstract class AbstractHadoopJob extends Configured implements Tool {
return totalMapInputMB;
}
+ protected double getTotalMapInputMB() throws ClassNotFoundException, IOException, InterruptedException, JobException {
+ return getTotalMapInputMB(job);
+ }
+
protected int getMapInputSplitCount() throws ClassNotFoundException, JobException, IOException, InterruptedException {
if (job == null) {
throw new JobException("Job is null");
http://git-wip-us.apache.org/repos/asf/kylin/blob/a38b02df/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 b2e186d..6a8ba4c 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
@@ -18,24 +18,19 @@
package org.apache.kylin.engine.mr.steps;
-import java.io.IOException;
-
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.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
-import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
-import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
-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.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.IMRInput.IMRTableInputFormat;
+import org.apache.kylin.engine.mr.IMROutput2;
import org.apache.kylin.engine.mr.MRUtil;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
@@ -84,11 +79,10 @@ public class CuboidJob extends AbstractHadoopJob {
options.addOption(OPTION_INPUT_PATH);
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 output = getOptionValue(OPTION_OUTPUT_PATH);
String cubeName = getOptionValue(OPTION_CUBE_NAME).toUpperCase();
int nCuboidLevel = Integer.parseInt(getOptionValue(OPTION_NCUBOID_LEVEL));
String segmentID = getOptionValue(OPTION_SEGMENT_ID);
@@ -109,8 +103,10 @@ public class CuboidJob extends AbstractHadoopJob {
setJobClasspath(job, cube.getConfig());
+ // add metadata to distributed cache
+ attachSegmentMetadataWithDict(segment, job.getConfiguration());
+
// Mapper
- configureMapperInputFormat(segment);
job.setMapperClass(this.mapperClass);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
@@ -118,22 +114,20 @@ public class CuboidJob extends AbstractHadoopJob {
// Reducer
job.setReducerClass(CuboidReducer.class);
- job.setOutputFormatClass(SequenceFileOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
- FileOutputFormat.setOutputPath(job, output);
+ // set input
+ configureMapperInputFormat(segment);
+
+ // set output
+ IMROutput2.IMROutputFormat outputFormat = MRUtil.getBatchCubingOutputSide2(segment).getOuputFormat();
+ outputFormat.configureJobOutput(job, output, segment, nCuboidLevel);
// set job configuration
job.getConfiguration().set(BatchConstants.CFG_CUBE_NAME, cubeName);
job.getConfiguration().set(BatchConstants.CFG_CUBE_SEGMENT_ID, segmentID);
job.getConfiguration().setInt(BatchConstants.CFG_CUBE_CUBOID_LEVEL, nCuboidLevel);
- // add metadata to distributed cache
- attachSegmentMetadataWithDict(segment, job.getConfiguration());
-
- job.setNumReduceTasks(LayerReducerNumSizing.getReduceTaskNum(segment, getTotalMapInputMB(), nCuboidLevel));
-
- this.deletePath(job.getConfiguration(), output);
return waitForCompletion(job);
} finally {
@@ -142,7 +136,7 @@ public class CuboidJob extends AbstractHadoopJob {
}
}
- private void configureMapperInputFormat(CubeSegment cubeSeg) throws IOException {
+ private void configureMapperInputFormat(CubeSegment cubeSeg) throws Exception {
String input = getOptionValue(OPTION_INPUT_PATH);
if ("FLAT_TABLE".equals(input)) {
@@ -151,12 +145,9 @@ public class CuboidJob extends AbstractHadoopJob {
flatTableInputFormat.configureJob(job);
} else {
// n-dimension cuboid case
+ IMROutput2.IMROutputFormat outputFormat = MRUtil.getBatchCubingOutputSide2(cubeSeg).getOuputFormat();
+ outputFormat.configureJobInput(job, input);
FileInputFormat.setInputPaths(job, new Path(input));
- if (hasOption(OPTION_INPUT_FORMAT) && ("textinputformat".equalsIgnoreCase(getOptionValue(OPTION_INPUT_FORMAT)))) {
- job.setInputFormatClass(TextInputFormat.class);
- } else {
- job.setInputFormatClass(SequenceFileInputFormat.class);
- }
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/a38b02df/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 7706bac..73a2eb9 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,29 +18,21 @@
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.Reducer.Context;
-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.common.util.HadoopUtil;
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.IMRInput.IMRTableInputFormat;
+import org.apache.kylin.engine.mr.IMROutput2;
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.execution.ExecutableManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -107,28 +99,24 @@ public class InMemCuboidJob extends AbstractHadoopJob {
job.getConfiguration().set(BatchConstants.CFG_CUBE_NAME, cubeName);
job.getConfiguration().set(BatchConstants.CFG_CUBE_SEGMENT_ID, segmentID);
- // set input
- IMRTableInputFormat flatTableInputFormat = MRUtil.getBatchCubingInputSide(segment).getFlatTableInputFormat();
- flatTableInputFormat.configureJob(job);
-
// set mapper
job.setMapperClass(InMemCuboidMapper.class);
job.setMapOutputKeyClass(ByteArrayWritable.class);
job.setMapOutputValueClass(ByteArrayWritable.class);
- // set output
- job.setReducerClass(InMemCuboidReducer.class);
- job.setNumReduceTasks(calculateReducerNum(segment));
-
+ // set reducer
// the cuboid file and KV class must be compatible with 0.7 version for smooth upgrade
- job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ job.setReducerClass(InMemCuboidReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
- Path outputPath = new Path(output);
- FileOutputFormat.setOutputPath(job, outputPath);
+ // set input
+ IMRTableInputFormat flatTableInputFormat = MRUtil.getBatchCubingInputSide(segment).getFlatTableInputFormat();
+ flatTableInputFormat.configureJob(job);
- HadoopUtil.deletePath(job.getConfiguration(), outputPath);
+ // set output
+ IMROutput2.IMROutputFormat outputFormat = MRUtil.getBatchCubingOutputSide2(segment).getOuputFormat();
+ outputFormat.configureJobOutput(job, output, segment, 0);
return waitForCompletion(job);
} finally {
@@ -137,31 +125,6 @@ public class InMemCuboidJob extends AbstractHadoopJob {
}
}
- private int calculateReducerNum(CubeSegment cubeSeg) throws IOException {
- KylinConfig kylinConfig = cubeSeg.getConfig();
-
- 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 by default
- numReduceTasks = Math.max(kylinConfig.getHadoopJobMinReducerNumber(), numReduceTasks);
- // no more than 500 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 " + Context.NUM_REDUCES + "=" + 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/a38b02df/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/LayerReducerNumSizing.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/LayerReducerNumSizing.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/LayerReducerNumSizing.java
deleted file mode 100644
index 7ce9842..0000000
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/LayerReducerNumSizing.java
+++ /dev/null
@@ -1,80 +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 org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.model.CubeDesc;
-import org.apache.kylin.engine.mr.common.CubeStatsReader;
-import org.apache.kylin.job.exception.JobException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class LayerReducerNumSizing {
-
- private static final Logger logger = LoggerFactory.getLogger(LayerReducerNumSizing.class);
-
- public static int getReduceTaskNum(CubeSegment cubeSegment, double totalMapInputMB, int level) throws ClassNotFoundException, IOException, InterruptedException, JobException {
- CubeDesc cubeDesc = cubeSegment.getCubeDesc();
- KylinConfig kylinConfig = cubeDesc.getConfig();
-
- double perReduceInputMB = kylinConfig.getDefaultHadoopJobReducerInputMB();
- double reduceCountRatio = kylinConfig.getDefaultHadoopJobReducerCountRatio();
- logger.info("Having per reduce MB " + perReduceInputMB + ", reduce count ratio " + reduceCountRatio + ", level " + level);
-
- CubeStatsReader cubeStatsReader = new CubeStatsReader(cubeSegment, kylinConfig);
-
- double parentLayerSizeEst, currentLayerSizeEst, adjustedCurrentLayerSizeEst;
-
- if (level == -1) {
- //merge case
- double estimatedSize = cubeStatsReader.estimateCubeSize();
- adjustedCurrentLayerSizeEst = estimatedSize > totalMapInputMB ? totalMapInputMB : estimatedSize;
- logger.info("estimated size {}, input size {}, adjustedCurrentLayerSizeEst: {}", estimatedSize, totalMapInputMB, adjustedCurrentLayerSizeEst);
- } else if (level == 0) {
- //base cuboid case TODO: the estimation could be very WRONG because it has no correction
- adjustedCurrentLayerSizeEst = cubeStatsReader.estimateLayerSize(0);
- logger.info("adjustedCurrentLayerSizeEst: {}", adjustedCurrentLayerSizeEst);
- } else {
- parentLayerSizeEst = cubeStatsReader.estimateLayerSize(level - 1);
- currentLayerSizeEst = cubeStatsReader.estimateLayerSize(level);
- adjustedCurrentLayerSizeEst = totalMapInputMB / parentLayerSizeEst * currentLayerSizeEst;
- logger.info("totalMapInputMB: {}, parentLayerSizeEst: {}, currentLayerSizeEst: {}, adjustedCurrentLayerSizeEst: {}", totalMapInputMB, parentLayerSizeEst, currentLayerSizeEst, adjustedCurrentLayerSizeEst);
- }
-
- // number of reduce tasks
- int numReduceTasks = (int) Math.round(adjustedCurrentLayerSizeEst / perReduceInputMB * reduceCountRatio + 0.99);
-
- // adjust reducer number for cube which has DISTINCT_COUNT measures for better performance
- if (cubeDesc.hasMemoryHungryMeasures()) {
- logger.info("Multiply reducer num by 4 to boost performance for memory hungry measures");
- numReduceTasks = numReduceTasks * 4;
- }
-
- // at least 1 reducer by default
- numReduceTasks = Math.max(kylinConfig.getHadoopJobMinReducerNumber(), numReduceTasks);
- // no more than 500 reducer by default
- numReduceTasks = Math.min(kylinConfig.getHadoopJobMaxReducerNumber(), numReduceTasks);
-
- return numReduceTasks;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/a38b02df/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidJob.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidJob.java
index 84b76e3..63d0619 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidJob.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidJob.java
@@ -19,15 +19,14 @@
package org.apache.kylin.engine.mr.steps;
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.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.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.engine.mr.IMROutput2;
+import org.apache.kylin.engine.mr.MRUtil;
import org.apache.kylin.engine.mr.common.BatchConstants;
public class MergeCuboidJob extends CuboidJob {
@@ -44,11 +43,14 @@ public class MergeCuboidJob extends CuboidJob {
options.addOption(OPTION_OUTPUT_PATH);
parseOptions(options, args);
+ String input = getOptionValue(OPTION_INPUT_PATH);
+ String output = getOptionValue(OPTION_OUTPUT_PATH);
String cubeName = getOptionValue(OPTION_CUBE_NAME).toUpperCase();
String segmentID = getOptionValue(OPTION_SEGMENT_ID);
CubeManager cubeMgr = CubeManager.getInstance(KylinConfig.getInstanceFromEnv());
CubeInstance cube = cubeMgr.getCube(cubeName);
+ CubeSegment cubeSeg = cube.getSegmentById(segmentID);
// start job
String jobName = getOptionValue(OPTION_JOB_NAME);
@@ -57,35 +59,32 @@ public class MergeCuboidJob extends CuboidJob {
setJobClasspath(job, cube.getConfig());
- // set inputs
- addInputDirs(getOptionValue(OPTION_INPUT_PATH), job);
-
- Path output = new Path(getOptionValue(OPTION_OUTPUT_PATH));
- FileOutputFormat.setOutputPath(job, output);
+ // add metadata to distributed cache
+ // TODO actually only dictionaries from merging segments are needed
+ attachCubeMetadataWithDict(cube, job.getConfiguration());
// Mapper
- job.setInputFormatClass(SequenceFileInputFormat.class);
job.setMapperClass(MergeCuboidMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
+ // Reducer
job.setReducerClass(CuboidReducer.class);
- job.setOutputFormatClass(SequenceFileOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
+ // set inputs
+ IMROutput2.IMRMergeOutputFormat outputFormat = MRUtil.getBatchMergeOutputSide2(cubeSeg).getOuputFormat();
+ outputFormat.configureJobInput(job, input);
+ addInputDirs(input, job);
+
+ // set output
+ outputFormat.configureJobOutput(job, output, cubeSeg);
+
// set job configuration
job.getConfiguration().set(BatchConstants.CFG_CUBE_NAME, cubeName);
job.getConfiguration().set(BatchConstants.CFG_CUBE_SEGMENT_ID, segmentID);
- // add metadata to distributed cache
- // TODO actually only dictionaries from merging segments are needed
- attachCubeMetadataWithDict(cube, job.getConfiguration());
-
- job.setNumReduceTasks(LayerReducerNumSizing.getReduceTaskNum(cube.getSegmentById(segmentID), getTotalMapInputMB(), -1));
-
- this.deletePath(job.getConfiguration(), output);
-
return waitForCompletion(job);
} finally {
if (job != null)
http://git-wip-us.apache.org/repos/asf/kylin/blob/a38b02df/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
index fccd48a..a603fc8 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
@@ -23,8 +23,6 @@ import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
@@ -44,7 +42,9 @@ 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.DictionaryManager;
+import org.apache.kylin.engine.mr.IMROutput2;
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.measure.BufferedMeasureCodec;
@@ -110,7 +110,8 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
// decide which source segment
FileSplit fileSplit = (FileSplit) context.getInputSplit();
- sourceCubeSegment = findSourceSegment(fileSplit, cube);
+ IMROutput2.IMRMergeOutputFormat outputFormat = MRUtil.getBatchMergeOutputSide2(mergedCubeSegment).getOuputFormat();
+ sourceCubeSegment = outputFormat.findSourceSegment(fileSplit, cube);
rowKeySplitter = new RowKeySplitter(sourceCubeSegment, 65, 255);
rowKeyEncoderProvider = new RowKeyEncoderProvider(mergedCubeSegment);
@@ -146,34 +147,6 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
}
}
- private static final Pattern JOB_NAME_PATTERN = Pattern.compile("kylin-([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})");
-
- public CubeSegment findSourceSegment(FileSplit fileSplit, CubeInstance cube) {
- String filePath = fileSplit.getPath().toString();
- String jobID = extractJobIDFromPath(filePath);
- return findSegmentWithUuid(jobID, cube);
- }
-
- private static String extractJobIDFromPath(String path) {
- Matcher matcher = JOB_NAME_PATTERN.matcher(path);
- // check the first occurrence
- if (matcher.find()) {
- return matcher.group(1);
- } else {
- throw new IllegalStateException("Can not extract job ID from file path : " + path);
- }
- }
-
- private static CubeSegment findSegmentWithUuid(String jobID, CubeInstance cubeInstance) {
- for (CubeSegment segment : cubeInstance.getSegments()) {
- String lastBuildJobID = segment.getLastBuildJobID();
- if (lastBuildJobID != null && lastBuildJobID.equalsIgnoreCase(jobID)) {
- return segment;
- }
- }
- throw new IllegalStateException("No merging segment's last build job ID equals " + jobID);
- }
-
@Override
public void doMap(Text key, Text value, Context context) throws IOException, InterruptedException {
long cuboidID = rowKeySplitter.split(key.getBytes());
http://git-wip-us.apache.org/repos/asf/kylin/blob/a38b02df/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/ReducerNumSizing.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/ReducerNumSizing.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/ReducerNumSizing.java
new file mode 100644
index 0000000..eb1adad
--- /dev/null
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/ReducerNumSizing.java
@@ -0,0 +1,106 @@
+/*
+ * 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.util.Map;
+
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.model.CubeDesc;
+import org.apache.kylin.engine.mr.common.CubeStatsReader;
+import org.apache.kylin.job.exception.JobException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ReducerNumSizing {
+
+ private static final Logger logger = LoggerFactory.getLogger(ReducerNumSizing.class);
+
+ public static int getLayeredCubingReduceTaskNum(CubeSegment cubeSegment, double totalMapInputMB, int level) throws ClassNotFoundException, IOException, InterruptedException, JobException {
+ CubeDesc cubeDesc = cubeSegment.getCubeDesc();
+ KylinConfig kylinConfig = cubeDesc.getConfig();
+
+ double perReduceInputMB = kylinConfig.getDefaultHadoopJobReducerInputMB();
+ double reduceCountRatio = kylinConfig.getDefaultHadoopJobReducerCountRatio();
+ logger.info("Having per reduce MB " + perReduceInputMB + ", reduce count ratio " + reduceCountRatio + ", level " + level);
+
+ CubeStatsReader cubeStatsReader = new CubeStatsReader(cubeSegment, kylinConfig);
+
+ double parentLayerSizeEst, currentLayerSizeEst, adjustedCurrentLayerSizeEst;
+
+ if (level == -1) {
+ //merge case
+ double estimatedSize = cubeStatsReader.estimateCubeSize();
+ adjustedCurrentLayerSizeEst = estimatedSize > totalMapInputMB ? totalMapInputMB : estimatedSize;
+ logger.info("estimated size {}, input size {}, adjustedCurrentLayerSizeEst: {}", estimatedSize, totalMapInputMB, adjustedCurrentLayerSizeEst);
+ } else if (level == 0) {
+ //base cuboid case TODO: the estimation could be very WRONG because it has no correction
+ adjustedCurrentLayerSizeEst = cubeStatsReader.estimateLayerSize(0);
+ logger.info("adjustedCurrentLayerSizeEst: {}", adjustedCurrentLayerSizeEst);
+ } else {
+ parentLayerSizeEst = cubeStatsReader.estimateLayerSize(level - 1);
+ currentLayerSizeEst = cubeStatsReader.estimateLayerSize(level);
+ adjustedCurrentLayerSizeEst = totalMapInputMB / parentLayerSizeEst * currentLayerSizeEst;
+ logger.info("totalMapInputMB: {}, parentLayerSizeEst: {}, currentLayerSizeEst: {}, adjustedCurrentLayerSizeEst: {}", totalMapInputMB, parentLayerSizeEst, currentLayerSizeEst, adjustedCurrentLayerSizeEst);
+ }
+
+ // number of reduce tasks
+ int numReduceTasks = (int) Math.round(adjustedCurrentLayerSizeEst / perReduceInputMB * reduceCountRatio + 0.99);
+
+ // adjust reducer number for cube which has DISTINCT_COUNT measures for better performance
+ if (cubeDesc.hasMemoryHungryMeasures()) {
+ logger.info("Multiply reducer num by 4 to boost performance for memory hungry measures");
+ numReduceTasks = numReduceTasks * 4;
+ }
+
+ // at least 1 reducer by default
+ numReduceTasks = Math.max(kylinConfig.getHadoopJobMinReducerNumber(), numReduceTasks);
+ // no more than 500 reducer by default
+ numReduceTasks = Math.min(kylinConfig.getHadoopJobMaxReducerNumber(), numReduceTasks);
+
+ return numReduceTasks;
+ }
+
+ public static int getInmemCubingReduceTaskNum(CubeSegment cubeSeg) throws IOException {
+ KylinConfig kylinConfig = cubeSeg.getConfig();
+
+ 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 by default
+ numReduceTasks = Math.max(kylinConfig.getHadoopJobMinReducerNumber(), numReduceTasks);
+ // no more than 500 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 " + Reducer.Context.NUM_REDUCES + "=" + numReduceTasks);
+ return numReduceTasks;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/a38b02df/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
----------------------------------------------------------------------
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
index d7a2c7e..6c542ab 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
@@ -112,8 +112,6 @@ public class HiveMRInput implements IMRInput {
HCatInputFormat.setInput(job, dbName, tableName);
job.setInputFormatClass(HCatInputFormat.class);
-
- job.setMapOutputValueClass(org.apache.hive.hcatalog.data.DefaultHCatRecord.class);
} catch (IOException e) {
throw new RuntimeException(e);
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/a38b02df/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
----------------------------------------------------------------------
diff --git a/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
index 500e1e9..4c140be 100644
--- a/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
+++ b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
@@ -117,7 +117,6 @@ public class KafkaMRInput implements IMRInput {
@Override
public void configureJob(Job job) {
job.setInputFormatClass(SequenceFileInputFormat.class);
- job.setMapOutputValueClass(Text.class);
String jobId = job.getConfiguration().get(BatchConstants.ARG_CUBING_JOB_ID);
IJoinedFlatTableDesc flatHiveTableDesc = new CubeJoinedFlatTableDesc(cubeSegment);
String inputPath = JoinedFlatTable.getTableDir(flatHiveTableDesc, JobBuilderSupport.getJobWorkingDir(conf, jobId));
http://git-wip-us.apache.org/repos/asf/kylin/blob/a38b02df/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 c4df354..31cb189 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
@@ -19,10 +19,25 @@
package org.apache.kylin.storage.hbase.steps;
import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.Job;
+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.util.HadoopUtil;
+import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.engine.mr.IMROutput2;
+import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
+import org.apache.kylin.engine.mr.steps.HiveToBaseCuboidMapper;
+import org.apache.kylin.engine.mr.steps.InMemCuboidMapper;
import org.apache.kylin.engine.mr.steps.MergeCuboidJob;
+import org.apache.kylin.engine.mr.steps.NDCuboidMapper;
+import org.apache.kylin.engine.mr.steps.ReducerNumSizing;
import org.apache.kylin.job.execution.DefaultChainedExecutable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -62,9 +77,38 @@ public class HBaseMROutput2Transition implements IMROutput2 {
public void addStepPhase4_Cleanup(DefaultChainedExecutable jobFlow) {
// nothing to do
}
+
+ @Override
+ public IMROutputFormat getOuputFormat() {
+ return new HBaseMROutputFormat();
+ }
};
}
+ public static class HBaseMROutputFormat implements IMROutputFormat {
+
+ @Override
+ public void configureJobInput(Job job, String input) throws Exception {
+ job.setInputFormatClass(SequenceFileInputFormat.class);
+ }
+
+ @Override
+ public void configureJobOutput(Job job, String output, CubeSegment segment, int level) throws Exception {
+ int reducerNum = 1;
+ Class mapperClass = job.getMapperClass();
+ if (mapperClass == HiveToBaseCuboidMapper.class || mapperClass == NDCuboidMapper.class) {
+ reducerNum = ReducerNumSizing.getLayeredCubingReduceTaskNum(segment, AbstractHadoopJob.getTotalMapInputMB(job), level);
+ } else if (mapperClass == InMemCuboidMapper.class) {
+ reducerNum = ReducerNumSizing.getInmemCubingReduceTaskNum(segment);
+ }
+ Path outputPath = new Path(output);
+ FileOutputFormat.setOutputPath(job, outputPath);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ job.setNumReduceTasks(reducerNum);
+ HadoopUtil.deletePath(job.getConfiguration(), outputPath);
+ }
+ }
+
@Override
public IMRBatchMergeOutputSide2 getBatchMergeOutputSide(final CubeSegment seg) {
return new IMRBatchMergeOutputSide2() {
@@ -86,6 +130,59 @@ public class HBaseMROutput2Transition implements IMROutput2 {
public void addStepPhase3_Cleanup(DefaultChainedExecutable jobFlow) {
steps.addMergingGarbageCollectionSteps(jobFlow);
}
+
+ @Override
+ public IMRMergeOutputFormat getOuputFormat() {
+ return new HBaseMergeMROutputFormat();
+ }
};
}
+
+ public static class HBaseMergeMROutputFormat implements IMRMergeOutputFormat{
+
+ private static final Pattern JOB_NAME_PATTERN = Pattern.compile("kylin-([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})");
+
+ @Override
+ public void configureJobInput(Job job, String input) throws Exception {
+ job.setInputFormatClass(SequenceFileInputFormat.class);
+ }
+
+ @Override
+ public void configureJobOutput(Job job, String output, CubeSegment segment) throws Exception {
+ int reducerNum = ReducerNumSizing.getLayeredCubingReduceTaskNum(segment, AbstractHadoopJob.getTotalMapInputMB(job), -1);
+ job.setNumReduceTasks(reducerNum);
+
+ Path outputPath = new Path(output);
+ HadoopUtil.deletePath(job.getConfiguration(), outputPath);
+ FileOutputFormat.setOutputPath(job, outputPath);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ }
+
+ @Override
+ public CubeSegment findSourceSegment(FileSplit fileSplit, CubeInstance cube) {
+ String filePath = fileSplit.getPath().toString();
+ String jobID = extractJobIDFromPath(filePath);
+ return findSegmentWithUuid(jobID, cube);
+ }
+
+ private static String extractJobIDFromPath(String path) {
+ Matcher matcher = JOB_NAME_PATTERN.matcher(path);
+ // check the first occurrence
+ if (matcher.find()) {
+ return matcher.group(1);
+ } else {
+ throw new IllegalStateException("Can not extract job ID from file path : " + path);
+ }
+ }
+
+ private static CubeSegment findSegmentWithUuid(String jobID, CubeInstance cubeInstance) {
+ for (CubeSegment segment : cubeInstance.getSegments()) {
+ String lastBuildJobID = segment.getLastBuildJobID();
+ if (lastBuildJobID != null && lastBuildJobID.equalsIgnoreCase(jobID)) {
+ return segment;
+ }
+ }
+ throw new IllegalStateException("No merging segment's last build job ID equals " + jobID);
+ }
+ }
}
\ No newline at end of file