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/07/10 10:39:05 UTC
[2/3] incubator-kylin git commit: KYLIN-876 Refactor CubingJobBuilder
into engine.mr package; Stop support of build+merge job
KYLIN-876 Refactor CubingJobBuilder into engine.mr package; Stop support
of build+merge job
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/87c32b0f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/87c32b0f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/87c32b0f
Branch: refs/heads/0.8
Commit: 87c32b0f74c67b93ce14ff7d9ad9b269f97a446d
Parents: cbf3550
Author: Li, Yang <ya...@ebay.com>
Authored: Fri Jul 10 10:16:37 2015 +0800
Committer: Li, Yang <ya...@ebay.com>
Committed: Fri Jul 10 16:36:55 2015 +0800
----------------------------------------------------------------------
.../org/apache/kylin/common/util/ClassUtil.java | 39 +-
.../apache/kylin/common/util/StringUtil.java | 10 +
.../apache/kylin/engine/BuildEngineFactory.java | 48 ++
.../apache/kylin/engine/IBatchCubeBuilder.java | 32 -
.../org/apache/kylin/engine/IBuildEngine.java | 12 +-
.../kylin/engine/mr/BatchCubingJobBuilder.java | 230 ++++++++
.../kylin/engine/mr/BatchMergeJobBuilder.java | 160 +++++
.../org/apache/kylin/engine/mr/CubingJob.java | 188 ++++++
.../kylin/engine/mr/GarbageCollectionStep.java | 148 +++++
.../org/apache/kylin/engine/mr/IMROutput.java | 1 +
.../kylin/engine/mr/JobBuilderSupport.java | 214 +++++++
.../apache/kylin/engine/mr/MRBuildEngine.java | 18 +-
.../apache/kylin/engine/mr/MRCubeBuilder.java | 45 --
.../kylin/engine/mr/MergeDictionaryStep.java | 197 +++++++
.../engine/mr/UpdateCubeInfoAfterBuildStep.java | 100 ++++
.../engine/mr/UpdateCubeInfoAfterMergeStep.java | 135 +++++
.../apache/kylin/job/AbstractJobBuilder.java | 26 -
.../apache/kylin/job/CubeMetadataUpgrade.java | 2 +-
.../org/apache/kylin/job/cube/CubingJob.java | 165 ------
.../apache/kylin/job/cube/CubingJobBuilder.java | 578 -------------------
.../kylin/job/cube/GarbageCollectionStep.java | 149 -----
.../kylin/job/cube/MergeDictionaryStep.java | 197 -------
.../job/cube/UpdateCubeInfoAfterBuildStep.java | 100 ----
.../job/cube/UpdateCubeInfoAfterMergeStep.java | 135 -----
.../kylin/job/engine/JobEngineConfig.java | 18 +-
.../kylin/job/invertedindex/IIJobBuilder.java | 37 +-
.../kylin/storage/hbase/HBaseMROutput.java | 6 +
.../kylin/job/BuildCubeWithEngineTest.java | 36 +-
.../apache/kylin/job/BuildIIWithEngineTest.java | 2 +-
.../apache/kylin/rest/service/BasicService.java | 2 +-
.../apache/kylin/rest/service/CubeService.java | 22 +-
.../apache/kylin/rest/service/JobService.java | 59 +-
32 files changed, 1583 insertions(+), 1528 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/common/src/main/java/org/apache/kylin/common/util/ClassUtil.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/kylin/common/util/ClassUtil.java b/common/src/main/java/org/apache/kylin/common/util/ClassUtil.java
index 63a06ed..38167a8 100644
--- a/common/src/main/java/org/apache/kylin/common/util/ClassUtil.java
+++ b/common/src/main/java/org/apache/kylin/common/util/ClassUtil.java
@@ -22,16 +22,14 @@ import java.io.File;
import java.lang.reflect.Method;
import java.net.URL;
import java.net.URLClassLoader;
+import java.util.HashMap;
+import java.util.Map;
import java.util.WeakHashMap;
/**
- * @author xduo
- *
*/
public class ClassUtil {
- private static WeakHashMap<String, Class<?>> forNameCache = new WeakHashMap<>();
-
public static void addClasspath(String path) throws Exception {
System.out.println("Adding path " + path + " to class path");
File file = new File(path);
@@ -45,21 +43,36 @@ public class ClassUtil {
}
}
+ private static final WeakHashMap<String, Class<?>> forNameCache = new WeakHashMap<>();
+ private static final Map<String, String> classRenameMap;
+ static {
+ classRenameMap = new HashMap<>();
+ classRenameMap.put("org.apache.kylin.job.cube.CubingJob", "org.apache.kylin.engine.mr.CubingJob");
+ classRenameMap.put("org.apache.kylin.job.cube.GarbageCollectionStep", "org.apache.kylin.engine.mr.GarbageCollectionStep");
+ classRenameMap.put("org.apache.kylin.job.cube.MergeDictionaryStep", "org.apache.kylin.engine.mr.MergeDictionaryStep");
+ classRenameMap.put("org.apache.kylin.job.cube.UpdateCubeInfoAfterBuildStep", "org.apache.kylin.engine.mr.UpdateCubeInfoAfterBuildStep");
+ classRenameMap.put("org.apache.kylin.job.cube.UpdateCubeInfoAfterMergeStep", "org.apache.kylin.engine.mr.UpdateCubeInfoAfterMergeStep");
+ }
+
@SuppressWarnings("unchecked")
public static <T> Class<? extends T> forName(String name, Class<T> clz) throws ClassNotFoundException {
+ String origName = name;
+
+ Class<? extends T> result = (Class<? extends T>) forNameCache.get(origName);
+ if (result == null) {
+ name = forRenamedClass(name);
+ result = (Class<? extends T>) Class.forName(name);
+ forNameCache.put(origName, result);
+ }
+ return result;
+ }
+ private static String forRenamedClass(String name) {
if (name.startsWith("com.kylinolap")) {
name = "org.apache.kylin" + name.substring("com.kylinolap".length());
}
-
- if (forNameCache.containsKey(name)) {
- return (Class<? extends T>) forNameCache.get(name);
- }
-
- Class<?> ret = Class.forName(name);
- forNameCache.put(name, ret);
-
- return (Class<? extends T>) ret;
+ String rename = classRenameMap.get(name);
+ return rename == null ? name : rename;
}
public static Object newInstance(String clz) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/common/src/main/java/org/apache/kylin/common/util/StringUtil.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/kylin/common/util/StringUtil.java b/common/src/main/java/org/apache/kylin/common/util/StringUtil.java
index 9cd5164..468a6e1 100644
--- a/common/src/main/java/org/apache/kylin/common/util/StringUtil.java
+++ b/common/src/main/java/org/apache/kylin/common/util/StringUtil.java
@@ -47,6 +47,16 @@ public class StringUtil {
return (String[]) whatsLeft.toArray(new String[whatsLeft.size()]);
}
+ public static String join(Iterable<String> parts, String separator) {
+ StringBuilder buf = new StringBuilder();
+ for (String p : parts) {
+ if (buf.length() > 0)
+ buf.append(separator);
+ buf.append(p);
+ }
+ return buf.toString();
+ }
+
public static void toUpperCaseArray(String[] source, String[] target) {
if(source!=null) {
for (int i = 0; i < source.length; i++) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/BuildEngineFactory.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/BuildEngineFactory.java b/job/src/main/java/org/apache/kylin/engine/BuildEngineFactory.java
new file mode 100644
index 0000000..7f3c761
--- /dev/null
+++ b/job/src/main/java/org/apache/kylin/engine/BuildEngineFactory.java
@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.engine.mr.MRBuildEngine;
+import org.apache.kylin.job.execution.DefaultChainedExecutable;
+
+public class BuildEngineFactory {
+
+ private static final IBuildEngine defaultEngine = new MRBuildEngine();
+
+ public static IBuildEngine getDefaultEngine() {
+ return defaultEngine;
+ }
+
+ public static Runnable createStreamingCubeBuilder(CubeSegment newSegment, String submitter) {
+ return getDefaultEngine().createStreamingCubeBuilder(newSegment, submitter);
+ }
+
+ /** Build a new cube segment, typically its time range appends to the end of current cube. */
+ public static DefaultChainedExecutable createBatchBuildJob(CubeSegment newSegment, String submitter) {
+ return getDefaultEngine().createBatchBuildJob(newSegment, submitter);
+ }
+
+ /** Merge multiple small segments into a big one. */
+ public static DefaultChainedExecutable createBatchMergeJob(CubeSegment mergeSegment, String submitter) {
+ return getDefaultEngine().createBatchMergeJob(mergeSegment, submitter);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/IBatchCubeBuilder.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/IBatchCubeBuilder.java b/job/src/main/java/org/apache/kylin/engine/IBatchCubeBuilder.java
deleted file mode 100644
index 2f5344a..0000000
--- a/job/src/main/java/org/apache/kylin/engine/IBatchCubeBuilder.java
+++ /dev/null
@@ -1,32 +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;
-
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.job.execution.ChainedExecutable;
-
-public interface IBatchCubeBuilder {
-
- /** Build a new cube segment, typically its time range appends to the end of current cube. */
- public ChainedExecutable createBuildJob(CubeSegment newSegment);
-
- /** Merge multiple small segments into a big one. */
- public ChainedExecutable createMergeJob(CubeSegment mergeSegment);
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/IBuildEngine.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/IBuildEngine.java b/job/src/main/java/org/apache/kylin/engine/IBuildEngine.java
index e54429f..909004c 100644
--- a/job/src/main/java/org/apache/kylin/engine/IBuildEngine.java
+++ b/job/src/main/java/org/apache/kylin/engine/IBuildEngine.java
@@ -18,7 +18,17 @@
package org.apache.kylin.engine;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.job.execution.DefaultChainedExecutable;
+
public interface IBuildEngine {
- public IBatchCubeBuilder createBatchCubeBuilder();
+ public Runnable createStreamingCubeBuilder(CubeSegment newSegment, String submitter);
+
+ /** Build a new cube segment, typically its time range appends to the end of current cube. */
+ public DefaultChainedExecutable createBatchBuildJob(CubeSegment newSegment, String submitter);
+
+ /** Merge multiple small segments into a big one. */
+ public DefaultChainedExecutable createBatchMergeJob(CubeSegment mergeSegment, String submitter);
+
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder.java b/job/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder.java
new file mode 100644
index 0000000..8f16775
--- /dev/null
+++ b/job/src/main/java/org/apache/kylin/engine/mr/BatchCubingJobBuilder.java
@@ -0,0 +1,230 @@
+/*
+ * 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;
+
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.model.CubeJoinedFlatTableDesc;
+import org.apache.kylin.job.common.HadoopShellExecutable;
+import org.apache.kylin.job.common.MapReduceExecutable;
+import org.apache.kylin.job.constant.ExecutableConstants;
+import org.apache.kylin.job.hadoop.cube.BaseCuboidJob;
+import org.apache.kylin.job.hadoop.cube.FactDistinctColumnsJob;
+import org.apache.kylin.job.hadoop.cube.NDCuboidJob;
+import org.apache.kylin.job.hadoop.cubev2.InMemCuboidJob;
+import org.apache.kylin.job.hadoop.cubev2.SaveStatisticsStep;
+import org.apache.kylin.job.hadoop.dict.CreateDictionaryJob;
+
+public class BatchCubingJobBuilder extends JobBuilderSupport {
+
+ public static interface IMRBuildInputParticipant {
+
+ }
+
+ public static interface IMRBuildOutputParticipant {
+
+ }
+
+ public BatchCubingJobBuilder(CubeSegment newSegment, String submitter) {
+ super(newSegment, submitter);
+ }
+
+ public CubingJob build() {
+ final CubingJob result = CubingJob.createBuildJob(seg, submitter, config);
+ final String jobId = result.getId();
+ final String cuboidRootPath = getCuboidRootPath(jobId);
+ final CubeJoinedFlatTableDesc flatHiveTableDesc = new CubeJoinedFlatTableDesc(seg.getCubeDesc(), seg);
+
+ result.addTask(createFlatHiveTableStep(flatHiveTableDesc, jobId));
+
+ result.addTask(createFactDistinctColumnsStep(flatHiveTableDesc, jobId));
+ result.addTask(createBuildDictionaryStep(jobId));
+
+ if (config.isInMemCubing()) {
+ result.addTask(createSaveStatisticsStep(jobId));
+
+ // create htable step
+ result.addTask(createCreateHTableStep(jobId));
+ result.addTask(createInMemCubingStep(flatHiveTableDesc, result.getId()));
+ // bulk load step
+ result.addTask(createBulkLoadStep(jobId));
+ } else {
+ final int groupRowkeyColumnsCount = seg.getCubeDesc().getRowkey().getNCuboidBuildLevels();
+ final int totalRowkeyColumnsCount = seg.getCubeDesc().getRowkey().getRowKeyColumns().length;
+ final String[] cuboidOutputTempPath = getCuboidOutputPaths(cuboidRootPath, totalRowkeyColumnsCount, groupRowkeyColumnsCount);
+
+ // base cuboid step
+ result.addTask(createBaseCuboidStep(flatHiveTableDesc, cuboidOutputTempPath, jobId));
+
+ // n dim cuboid steps
+ for (int i = 1; i <= groupRowkeyColumnsCount; i++) {
+ int dimNum = totalRowkeyColumnsCount - i;
+ result.addTask(createNDimensionCuboidStep(cuboidOutputTempPath, dimNum, totalRowkeyColumnsCount));
+ }
+ result.addTask(createRangeRowkeyDistributionStep(cuboidRootPath + "*", jobId));
+ // create htable step
+ result.addTask(createCreateHTableStep(jobId));
+ // generate hfiles step
+ result.addTask(createConvertCuboidToHfileStep(cuboidRootPath + "*", jobId));
+ // bulk load step
+ result.addTask(createBulkLoadStep(jobId));
+ }
+
+ result.addTask(createUpdateCubeInfoAfterBuildStep(jobId));
+
+ result.addTask(createGarbageCollectionStep(null, flatHiveTableDesc.getTableName(jobId)));
+
+ return result;
+ }
+
+ private MapReduceExecutable createFactDistinctColumnsStep(CubeJoinedFlatTableDesc flatHiveTableDesc, String jobId) {
+ MapReduceExecutable result = new MapReduceExecutable();
+ result.setName(ExecutableConstants.STEP_NAME_FACT_DISTINCT_COLUMNS);
+ result.setMapReduceJobClass(FactDistinctColumnsJob.class);
+ StringBuilder cmd = new StringBuilder();
+ appendMapReduceParameters(cmd, seg);
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
+ appendExecCmdParameters(cmd, "output", getFactDistinctColumnsPath(jobId));
+ appendExecCmdParameters(cmd, "segmentname", seg.getName());
+ appendExecCmdParameters(cmd, "statisticsenabled", String.valueOf(config.isInMemCubing()));
+ appendExecCmdParameters(cmd, "statisticsoutput", getStatisticsPath(jobId));
+ appendExecCmdParameters(cmd, "statisticssamplingpercent", String.valueOf(config.getConfig().getCubingInMemSamplingPercent()));
+ appendExecCmdParameters(cmd, "jobname", "Kylin_Fact_Distinct_Columns_" + seg.getCubeInstance().getName() + "_Step");
+ appendExecCmdParameters(cmd, "tablename", flatHiveTableDesc.getTableName(jobId));
+
+ result.setMapReduceParams(cmd.toString());
+ return result;
+ }
+
+ private HadoopShellExecutable createBuildDictionaryStep(String jobId) {
+ // base cuboid job
+ HadoopShellExecutable buildDictionaryStep = new HadoopShellExecutable();
+ buildDictionaryStep.setName(ExecutableConstants.STEP_NAME_BUILD_DICTIONARY);
+ StringBuilder cmd = new StringBuilder();
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
+ appendExecCmdParameters(cmd, "segmentname", seg.getName());
+ appendExecCmdParameters(cmd, "input", getFactDistinctColumnsPath(jobId));
+
+ buildDictionaryStep.setJobParams(cmd.toString());
+ buildDictionaryStep.setJobClass(CreateDictionaryJob.class);
+ return buildDictionaryStep;
+ }
+
+ private MapReduceExecutable createBaseCuboidStep(CubeJoinedFlatTableDesc flatHiveTableDesc, String[] cuboidOutputTempPath, String jobId) {
+ // base cuboid job
+ MapReduceExecutable baseCuboidStep = new MapReduceExecutable();
+
+ StringBuilder cmd = new StringBuilder();
+ appendMapReduceParameters(cmd, seg);
+
+ baseCuboidStep.setName(ExecutableConstants.STEP_NAME_BUILD_BASE_CUBOID);
+
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
+ appendExecCmdParameters(cmd, "segmentname", seg.getName());
+ appendExecCmdParameters(cmd, "input", getFlatHiveTableLocation(flatHiveTableDesc, jobId));
+ appendExecCmdParameters(cmd, "output", cuboidOutputTempPath[0]);
+ appendExecCmdParameters(cmd, "jobname", "Kylin_Base_Cuboid_Builder_" + seg.getCubeInstance().getName());
+ appendExecCmdParameters(cmd, "level", "0");
+
+ 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) {
+ // ND cuboid job
+ MapReduceExecutable ndCuboidStep = new MapReduceExecutable();
+
+ ndCuboidStep.setName(ExecutableConstants.STEP_NAME_BUILD_N_D_CUBOID + " : " + dimNum + "-Dimension");
+ StringBuilder cmd = new StringBuilder();
+
+ appendMapReduceParameters(cmd, seg);
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().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.getCubeInstance().getName() + "_Step");
+ appendExecCmdParameters(cmd, "level", "" + (totalRowkeyColumnCount - dimNum));
+
+ ndCuboidStep.setMapReduceParams(cmd.toString());
+ ndCuboidStep.setMapReduceJobClass(NDCuboidJob.class);
+ return ndCuboidStep;
+ }
+
+ private SaveStatisticsStep createSaveStatisticsStep(String jobId) {
+ SaveStatisticsStep result = new SaveStatisticsStep();
+ result.setName(ExecutableConstants.STEP_NAME_SAVE_STATISTICS);
+ result.setCubeName(seg.getCubeInstance().getName());
+ result.setSegmentId(seg.getUuid());
+ result.setStatisticsPath(getStatisticsPath(jobId));
+ return result;
+ }
+
+ private MapReduceExecutable createInMemCubingStep(CubeJoinedFlatTableDesc flatHiveTableDesc, String jobId) {
+ // base cuboid job
+ MapReduceExecutable baseCuboidStep = new MapReduceExecutable();
+
+ StringBuilder cmd = new StringBuilder();
+ appendMapReduceParameters(cmd, seg);
+
+ baseCuboidStep.setName(ExecutableConstants.STEP_NAME_BUILD_IN_MEM_CUBE);
+
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
+ appendExecCmdParameters(cmd, "segmentname", seg.getName());
+ appendExecCmdParameters(cmd, "input", getFlatHiveTableLocation(flatHiveTableDesc, jobId));
+ appendExecCmdParameters(cmd, "statisticsoutput", getStatisticsPath(jobId));
+ appendExecCmdParameters(cmd, "output", getHFilePath(jobId));
+ appendExecCmdParameters(cmd, "jobname", "Kylin_Cube_Builder_" + seg.getCubeInstance().getName());
+ appendExecCmdParameters(cmd, "level", "0");
+ appendExecCmdParameters(cmd, "tablename", flatHiveTableDesc.getTableName(jobId));
+ appendExecCmdParameters(cmd, "htablename", seg.getStorageLocationIdentifier());
+
+ baseCuboidStep.setMapReduceParams(cmd.toString());
+ baseCuboidStep.setMapReduceJobClass(InMemCuboidJob.class);
+ baseCuboidStep.setCounterSaveAs(CubingJob.SOURCE_RECORD_COUNT + "," + CubingJob.SOURCE_SIZE_BYTES + "," + CubingJob.CUBE_SIZE_BYTES);
+ return baseCuboidStep;
+ }
+
+ private UpdateCubeInfoAfterBuildStep createUpdateCubeInfoAfterBuildStep(String jobId) {
+ final UpdateCubeInfoAfterBuildStep updateCubeInfoStep = new UpdateCubeInfoAfterBuildStep();
+ updateCubeInfoStep.setName(ExecutableConstants.STEP_NAME_UPDATE_CUBE_INFO);
+ updateCubeInfoStep.setCubeName(seg.getCubeInstance().getName());
+ updateCubeInfoStep.setSegmentId(seg.getUuid());
+ updateCubeInfoStep.setCubingJobId(jobId);
+ return updateCubeInfoStep;
+ }
+
+ private String getFlatHiveTableLocation(CubeJoinedFlatTableDesc flatTableDesc, String jobId) {
+ return getJobWorkingDir(jobId) + "/" + flatTableDesc.getTableName(jobId);
+ }
+
+ 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/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder.java b/job/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder.java
new file mode 100644
index 0000000..cb2d8dd
--- /dev/null
+++ b/job/src/main/java/org/apache/kylin/engine/mr/BatchMergeJobBuilder.java
@@ -0,0 +1,160 @@
+/*
+ * 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;
+
+import java.util.List;
+
+import org.apache.kylin.common.util.StringUtil;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.job.common.MapReduceExecutable;
+import org.apache.kylin.job.constant.ExecutableConstants;
+import org.apache.kylin.job.hadoop.cube.MergeCuboidJob;
+import org.apache.kylin.job.hadoop.cubev2.MergeCuboidFromHBaseJob;
+import org.apache.kylin.job.hadoop.cubev2.MergeStatisticsStep;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+public class BatchMergeJobBuilder extends JobBuilderSupport {
+
+ public BatchMergeJobBuilder(CubeSegment mergeSegment, String submitter) {
+ super(mergeSegment, submitter);
+ }
+
+ public CubingJob build() {
+ final CubingJob result = CubingJob.createMergeJob(seg, submitter, config);
+ final String jobId = result.getId();
+ final String cuboidRootPath = getCuboidRootPath(jobId);
+
+ final List<CubeSegment> mergingSegments = seg.getCubeInstance().getMergingSegments(seg);
+ Preconditions.checkState(mergingSegments.size() > 1, "there should be more than 2 segments to merge");
+ final List<String> mergingSegmentIds = Lists.newArrayList();
+ final List<String> mergingCuboidPaths = Lists.newArrayList();
+ final List<String> mergingHTables = Lists.newArrayList();
+ for (CubeSegment merging : mergingSegments) {
+ mergingSegmentIds.add(merging.getUuid());
+ mergingCuboidPaths.add(getCuboidRootPath(merging) + "*");
+ mergingHTables.add(merging.getStorageLocationIdentifier());
+ }
+
+ result.addTask(createMergeDictionaryStep(mergingSegmentIds));
+
+ if (config.isInMemCubing()) {
+
+ String mergedStatisticsFolder = getStatisticsPath(jobId);
+ result.addTask(createMergeStatisticsStep(seg, mergingSegmentIds, mergedStatisticsFolder));
+
+ // create htable step
+ result.addTask(createCreateHTableStep(jobId));
+
+ String formattedTables = StringUtil.join(mergingHTables, ",");
+ result.addTask(createMergeCuboidDataFromHBaseStep(formattedTables, jobId));
+
+ } else {
+ // merge cuboid
+ String formattedPath = StringUtil.join(mergingCuboidPaths, ",");
+ result.addTask(createMergeCuboidDataStep(seg, formattedPath, cuboidRootPath));
+
+ // convert htable
+ result.addTask(createRangeRowkeyDistributionStep(cuboidRootPath + "*", jobId));
+ // create htable step
+ result.addTask(createCreateHTableStep(jobId));
+ // generate hfiles step
+ result.addTask(createConvertCuboidToHfileStep(cuboidRootPath + "*", jobId));
+ }
+
+ // bulk load step
+ result.addTask(createBulkLoadStep(jobId));
+
+ // update cube info
+ result.addTask(createUpdateCubeInfoAfterMergeStep(mergingSegmentIds, jobId));
+
+ result.addTask(createGarbageCollectionStep(mergingHTables, null));
+
+ return result;
+ }
+
+ private MergeDictionaryStep createMergeDictionaryStep(List<String> mergingSegmentIds) {
+ MergeDictionaryStep result = new MergeDictionaryStep();
+ result.setName(ExecutableConstants.STEP_NAME_MERGE_DICTIONARY);
+ result.setCubeName(seg.getCubeInstance().getName());
+ result.setSegmentId(seg.getUuid());
+ result.setMergingSegmentIds(mergingSegmentIds);
+ return result;
+ }
+
+ private MergeStatisticsStep createMergeStatisticsStep(CubeSegment seg, List<String> mergingSegmentIds, String mergedStatisticsFolder) {
+ MergeStatisticsStep result = new MergeStatisticsStep();
+ result.setName(ExecutableConstants.STEP_NAME_MERGE_STATISTICS);
+ result.setCubeName(seg.getCubeInstance().getName());
+ result.setSegmentId(seg.getUuid());
+ result.setMergingSegmentIds(mergingSegmentIds);
+ result.setMergedStatisticsPath(mergedStatisticsFolder);
+ return result;
+ }
+
+ 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, seg);
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
+ appendExecCmdParameters(cmd, "segmentname", seg.getName());
+ appendExecCmdParameters(cmd, "input", inputPath);
+ appendExecCmdParameters(cmd, "output", outputPath);
+ appendExecCmdParameters(cmd, "jobname", "Kylin_Merge_Cuboid_" + seg.getCubeInstance().getName() + "_Step");
+
+ mergeCuboidDataStep.setMapReduceParams(cmd.toString());
+ mergeCuboidDataStep.setMapReduceJobClass(MergeCuboidJob.class);
+ return mergeCuboidDataStep;
+ }
+
+
+ private MapReduceExecutable createMergeCuboidDataFromHBaseStep(String inputTableNames, String jobId) {
+ MapReduceExecutable mergeCuboidDataStep = new MapReduceExecutable();
+ mergeCuboidDataStep.setName(ExecutableConstants.STEP_NAME_MERGE_CUBOID);
+ StringBuilder cmd = new StringBuilder();
+
+ appendMapReduceParameters(cmd, seg);
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
+ appendExecCmdParameters(cmd, "segmentname", seg.getName());
+ appendExecCmdParameters(cmd, "input", inputTableNames);
+ appendExecCmdParameters(cmd, "output", getHFilePath(jobId));
+ appendExecCmdParameters(cmd, "htablename", seg.getStorageLocationIdentifier());
+ appendExecCmdParameters(cmd, "jobname", "Kylin_Merge_Cuboid_" + seg.getCubeInstance().getName() + "_Step");
+
+ mergeCuboidDataStep.setMapReduceParams(cmd.toString());
+ mergeCuboidDataStep.setMapReduceJobClass(MergeCuboidFromHBaseJob.class);
+ mergeCuboidDataStep.setCounterSaveAs(",," + CubingJob.CUBE_SIZE_BYTES);
+ return mergeCuboidDataStep;
+ }
+
+ private UpdateCubeInfoAfterMergeStep createUpdateCubeInfoAfterMergeStep(List<String> mergingSegmentIds, String jobId) {
+ UpdateCubeInfoAfterMergeStep result = new UpdateCubeInfoAfterMergeStep();
+ result.setName(ExecutableConstants.STEP_NAME_UPDATE_CUBE_INFO);
+ result.setCubeName(seg.getCubeInstance().getName());
+ result.setSegmentId(seg.getUuid());
+ result.setMergingSegmentIds(mergingSegmentIds);
+ result.setCubingJobId(jobId);
+ return result;
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/mr/CubingJob.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/mr/CubingJob.java b/job/src/main/java/org/apache/kylin/engine/mr/CubingJob.java
new file mode 100644
index 0000000..77cbab7
--- /dev/null
+++ b/job/src/main/java/org/apache/kylin/engine/mr/CubingJob.java
@@ -0,0 +1,188 @@
+/*
+ * 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;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.TimeZone;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.kylin.cube.CubeInstance;
+import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.job.common.MapReduceExecutable;
+import org.apache.kylin.job.constant.ExecutableConstants;
+import org.apache.kylin.job.engine.JobEngineConfig;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.DefaultChainedExecutable;
+import org.apache.kylin.job.execution.ExecutableContext;
+import org.apache.kylin.job.execution.ExecutableState;
+import org.apache.kylin.job.execution.ExecuteResult;
+import org.apache.kylin.job.execution.Output;
+
+/**
+ */
+public class CubingJob extends DefaultChainedExecutable {
+
+ // KEYS of Output.extraInfo map, info passed across job steps
+ public static final String SOURCE_RECORD_COUNT = "sourceRecordCount";
+ public static final String SOURCE_SIZE_BYTES = "sourceSizeBytes";
+ public static final String CUBE_SIZE_BYTES = "byteSizeBytes";
+ public static final String MAP_REDUCE_WAIT_TIME = "mapReduceWaitTime";
+
+ private static final String CUBE_INSTANCE_NAME = "cubeName";
+ private static final String SEGMENT_ID = "segmentId";
+
+ public static CubingJob createBuildJob(CubeSegment seg, String submitter, JobEngineConfig config) {
+ return initCubingJob(seg, "BUILD", submitter, config);
+ }
+
+ public static CubingJob createMergeJob(CubeSegment seg, String submitter, JobEngineConfig config) {
+ return initCubingJob(seg, "MERGE", submitter, config);
+ }
+
+ private static CubingJob initCubingJob(CubeSegment seg, String jobType, String submitter, JobEngineConfig config) {
+ CubingJob result = new CubingJob();
+ SimpleDateFormat format = new SimpleDateFormat("z yyyy-MM-dd HH:mm:ss");
+ format.setTimeZone(TimeZone.getTimeZone(config.getTimeZone()));
+ result.setCubeName(seg.getCubeInstance().getName());
+ result.setSegmentId(seg.getUuid());
+ result.setName(seg.getCubeInstance().getName() + " - " + seg.getName() + " - " + jobType + " - " + format.format(new Date(System.currentTimeMillis())));
+ result.setSubmitter(submitter);
+ result.setNotifyList(seg.getCubeInstance().getDescriptor().getNotifyList());
+ return result;
+ }
+
+ public CubingJob() {
+ super();
+ }
+
+ void setCubeName(String name) {
+ setParam(CUBE_INSTANCE_NAME, name);
+ }
+
+ public String getCubeName() {
+ return getParam(CUBE_INSTANCE_NAME);
+ }
+
+ void setSegmentId(String segmentId) {
+ setParam(SEGMENT_ID, segmentId);
+ }
+
+ public String getSegmentId() {
+ return getParam(SEGMENT_ID);
+ }
+
+ @Override
+ protected Pair<String, String> formatNotifications(ExecutableContext context, ExecutableState state) {
+ CubeInstance cubeInstance = CubeManager.getInstance(context.getConfig()).getCube(getCubeName());
+ final Output output = jobService.getOutput(getId());
+ String logMsg;
+ state = output.getState();
+ if (state != ExecutableState.ERROR &&
+ !cubeInstance.getDescriptor().getStatusNeedNotify().contains(state.toString().toLowerCase())) {
+ logger.info("state:" + state + " no need to notify users");
+ return null;
+ }
+ switch (state) {
+ case ERROR:
+ logMsg = output.getVerboseMsg();
+ break;
+ case DISCARDED:
+ logMsg = "job has been discarded";
+ break;
+ case SUCCEED:
+ logMsg = "job has succeeded";
+ break;
+ default:
+ return null;
+ }
+ if (logMsg == null) {
+ logMsg = "no error message";
+ }
+ String content = ExecutableConstants.NOTIFY_EMAIL_TEMPLATE;
+ content = content.replaceAll("\\$\\{job_name\\}", getName());
+ content = content.replaceAll("\\$\\{result\\}", state.toString());
+ content = content.replaceAll("\\$\\{cube_name\\}", getCubeName());
+ content = content.replaceAll("\\$\\{start_time\\}", new Date(getStartTime()).toString());
+ content = content.replaceAll("\\$\\{duration\\}", getDuration() / 60000 + "mins");
+ content = content.replaceAll("\\$\\{mr_waiting\\}", getMapReduceWaitTime() / 60000 + "mins");
+ content = content.replaceAll("\\$\\{last_update_time\\}", new Date(getLastModified()).toString());
+ content = content.replaceAll("\\$\\{submitter\\}", getSubmitter());
+ content = content.replaceAll("\\$\\{error_log\\}", logMsg);
+
+ try {
+ InetAddress inetAddress = InetAddress.getLocalHost();
+ content = content.replaceAll("\\$\\{job_engine\\}", inetAddress.getCanonicalHostName());
+ } catch (UnknownHostException e) {
+ logger.warn(e.getLocalizedMessage(), e);
+ }
+
+ String title = "["+ state.toString() + "] - [Kylin Cube Build Job]-" + getCubeName();
+ return Pair.of(title, content);
+ }
+
+ @Override
+ protected void onExecuteFinished(ExecuteResult result, ExecutableContext executableContext) {
+ long time = 0L;
+ for (AbstractExecutable task: getTasks()) {
+ final ExecutableState status = task.getStatus();
+ if (status != ExecutableState.SUCCEED) {
+ break;
+ }
+ if (task instanceof MapReduceExecutable) {
+ time += ((MapReduceExecutable) task).getMapReduceWaitTime();
+ }
+ }
+ setMapReduceWaitTime(time);
+ super.onExecuteFinished(result, executableContext);
+ }
+
+ public long getMapReduceWaitTime() {
+ return getExtraInfoAsLong(MAP_REDUCE_WAIT_TIME, 0L);
+ }
+
+ public void setMapReduceWaitTime(long t) {
+ addExtraInfo(MAP_REDUCE_WAIT_TIME, t + "");
+ }
+
+ public long findSourceRecordCount() {
+ return Long.parseLong(findExtraInfo(SOURCE_RECORD_COUNT, "0"));
+ }
+
+ public long findSourceSizeBytes() {
+ return Long.parseLong(findExtraInfo(SOURCE_SIZE_BYTES, "0"));
+ }
+
+ public long findCubeSizeBytes() {
+ return Long.parseLong(findExtraInfo(CUBE_SIZE_BYTES, "0"));
+ }
+
+ private String findExtraInfo(String key, String dft) {
+ for (AbstractExecutable child : getTasks()) {
+ Output output = executableManager.getOutput(child.getId());
+ String value = output.getExtra().get(key);
+ if (value != null)
+ return value;
+ }
+ return dft;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/mr/GarbageCollectionStep.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/mr/GarbageCollectionStep.java b/job/src/main/java/org/apache/kylin/engine/mr/GarbageCollectionStep.java
new file mode 100644
index 0000000..2f7bf21
--- /dev/null
+++ b/job/src/main/java/org/apache/kylin/engine/mr/GarbageCollectionStep.java
@@ -0,0 +1,148 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.Bytes;
+import org.apache.kylin.job.cmd.ShellCmdOutput;
+import org.apache.kylin.job.exception.ExecuteException;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.ExecutableContext;
+import org.apache.kylin.job.execution.ExecuteResult;
+import org.apache.kylin.metadata.realization.IRealizationConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Drop the resources that is no longer needed, including intermediate hive table (after cube build) and hbase tables (after cube merge)
+ */
+public class GarbageCollectionStep extends AbstractExecutable {
+
+ private static final String OLD_HTABLES = "oldHTables";
+
+ private static final String OLD_HIVE_TABLE = "oldHiveTable";
+
+ private static final Logger logger = LoggerFactory.getLogger(GarbageCollectionStep.class);
+
+ public GarbageCollectionStep() {
+ super();
+ }
+
+ @Override
+ protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException {
+
+ StringBuffer output = new StringBuffer();
+
+ final String hiveTable = this.getOldHiveTable();
+ if (StringUtils.isNotEmpty(hiveTable)) {
+ final String dropHiveCMD = "hive -e \"DROP TABLE IF EXISTS " + hiveTable + ";\"";
+ ShellCmdOutput shellCmdOutput = new ShellCmdOutput();
+ try {
+ context.getConfig().getCliCommandExecutor().execute(dropHiveCMD, shellCmdOutput);
+ output.append("Hive table " + hiveTable + " is dropped. \n");
+ } catch (IOException e) {
+ logger.error("job:" + getId() + " execute finished with exception", e);
+ output.append(shellCmdOutput.getOutput()).append("\n").append(e.getLocalizedMessage());
+ return new ExecuteResult(ExecuteResult.State.ERROR, output.toString());
+ }
+ }
+
+
+ List<String> oldTables = getOldHTables();
+ if (oldTables != null && oldTables.size() > 0) {
+ String metadataUrlPrefix = KylinConfig.getInstanceFromEnv().getMetadataUrlPrefix();
+ Configuration conf = HBaseConfiguration.create();
+ HBaseAdmin admin = null;
+ try {
+ admin = new HBaseAdmin(conf);
+ for (String table : oldTables) {
+ if (admin.tableExists(table)) {
+ HTableDescriptor tableDescriptor = admin.getTableDescriptor(Bytes.toBytes(table));
+ String host = tableDescriptor.getValue(IRealizationConstants.HTableTag);
+ if (metadataUrlPrefix.equalsIgnoreCase(host)) {
+ if (admin.isTableEnabled(table)) {
+ admin.disableTable(table);
+ }
+ admin.deleteTable(table);
+ logger.debug("Dropped htable: " + table);
+ output.append("HBase table " + table + " is dropped. \n");
+ } else {
+ logger.debug("Skip htable: " + table);
+ output.append("Skip htable: " + table + ". \n");
+ }
+ }
+ }
+
+ } catch (IOException e) {
+ output.append("Got error when drop HBase table, exiting... \n");
+ // This should not block the merge job; Orphans should be cleaned up in StorageCleanupJob
+ return new ExecuteResult(ExecuteResult.State.ERROR, output.append(e.getLocalizedMessage()).toString());
+ } finally {
+ if (admin != null)
+ try {
+ admin.close();
+ } catch (IOException e) {
+ logger.error(e.getLocalizedMessage());
+ }
+ }
+ }
+
+
+ return new ExecuteResult(ExecuteResult.State.SUCCEED, output.toString());
+ }
+
+ public void setOldHTables(List<String> ids) {
+ setParam(OLD_HTABLES, StringUtils.join(ids, ","));
+ }
+
+ private List<String> getOldHTables() {
+ final String ids = getParam(OLD_HTABLES);
+ if (ids != null) {
+ final String[] splitted = StringUtils.split(ids, ",");
+ ArrayList<String> result = Lists.newArrayListWithExpectedSize(splitted.length);
+ for (String id : splitted) {
+ result.add(id);
+ }
+ return result;
+ } else {
+ return Collections.emptyList();
+ }
+ }
+
+ public void setOldHiveTable(String hiveTable) {
+ setParam(OLD_HIVE_TABLE, hiveTable);
+ }
+
+ private String getOldHiveTable() {
+ return getParam(OLD_HIVE_TABLE);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/mr/IMROutput.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/mr/IMROutput.java b/job/src/main/java/org/apache/kylin/engine/mr/IMROutput.java
index 697957e..6e3a42c 100644
--- a/job/src/main/java/org/apache/kylin/engine/mr/IMROutput.java
+++ b/job/src/main/java/org/apache/kylin/engine/mr/IMROutput.java
@@ -22,4 +22,5 @@ public interface IMROutput {
public IMRJobFlowParticipant createBuildFlowParticipant();
+ public IMRJobFlowParticipant createMergeFlowParticipant();
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java b/job/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java
new file mode 100644
index 0000000..b1f3c4e
--- /dev/null
+++ b/job/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java
@@ -0,0 +1,214 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.job.JoinedFlatTable;
+import org.apache.kylin.job.common.HadoopShellExecutable;
+import org.apache.kylin.job.common.MapReduceExecutable;
+import org.apache.kylin.job.common.ShellExecutable;
+import org.apache.kylin.job.constant.ExecutableConstants;
+import org.apache.kylin.job.engine.JobEngineConfig;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.hadoop.cube.CubeHFileJob;
+import org.apache.kylin.job.hadoop.cube.RangeKeyDistributionJob;
+import org.apache.kylin.job.hadoop.hbase.BulkLoadJob;
+import org.apache.kylin.job.hadoop.hbase.CreateHTableJob;
+import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Hold reusable methods for real builders.
+ */
+abstract public class JobBuilderSupport {
+
+ protected final JobEngineConfig config;
+ protected final CubeSegment seg;
+ protected final String submitter;
+
+ protected JobBuilderSupport(CubeSegment seg, String submitter) {
+ Preconditions.checkNotNull(seg, "segment cannot be null");
+ this.config = new JobEngineConfig(KylinConfig.getInstanceFromEnv());
+ this.seg = seg;
+ this.submitter = submitter;
+ }
+
+ protected AbstractExecutable createFlatHiveTableStep(IJoinedFlatTableDesc flatTableDesc, String jobId) {
+ return createFlatHiveTableStep(config, flatTableDesc, jobId);
+ }
+
+ protected MapReduceExecutable createRangeRowkeyDistributionStep(String inputPath, String jobId) {
+ MapReduceExecutable rowkeyDistributionStep = new MapReduceExecutable();
+ rowkeyDistributionStep.setName(ExecutableConstants.STEP_NAME_GET_CUBOID_KEY_DISTRIBUTION);
+ StringBuilder cmd = new StringBuilder();
+
+ appendMapReduceParameters(cmd, seg);
+ appendExecCmdParameters(cmd, "input", inputPath);
+ appendExecCmdParameters(cmd, "output", getRowkeyDistributionOutputPath(jobId));
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
+ appendExecCmdParameters(cmd, "jobname", "Kylin_Region_Splits_Calculator_" + seg.getCubeInstance().getName() + "_Step");
+
+ rowkeyDistributionStep.setMapReduceParams(cmd.toString());
+ rowkeyDistributionStep.setMapReduceJobClass(RangeKeyDistributionJob.class);
+ return rowkeyDistributionStep;
+ }
+
+ protected HadoopShellExecutable createCreateHTableStep(String jobId) {
+ HadoopShellExecutable createHtableStep = new HadoopShellExecutable();
+ createHtableStep.setName(ExecutableConstants.STEP_NAME_CREATE_HBASE_TABLE);
+ StringBuilder cmd = new StringBuilder();
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
+ appendExecCmdParameters(cmd, "segmentname", seg.getName());
+ appendExecCmdParameters(cmd, "input", getRowkeyDistributionOutputPath(jobId) + "/part-r-00000");
+ appendExecCmdParameters(cmd, "htablename", seg.getStorageLocationIdentifier());
+ appendExecCmdParameters(cmd, "statisticsenabled", String.valueOf(config.isInMemCubing()));
+
+ createHtableStep.setJobParams(cmd.toString());
+ createHtableStep.setJobClass(CreateHTableJob.class);
+
+ return createHtableStep;
+ }
+
+ protected MapReduceExecutable createConvertCuboidToHfileStep(String inputPath, String jobId) {
+ MapReduceExecutable createHFilesStep = new MapReduceExecutable();
+ createHFilesStep.setName(ExecutableConstants.STEP_NAME_CONVERT_CUBOID_TO_HFILE);
+ StringBuilder cmd = new StringBuilder();
+
+ appendMapReduceParameters(cmd, seg);
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
+ appendExecCmdParameters(cmd, "input", inputPath);
+ appendExecCmdParameters(cmd, "output", getHFilePath(jobId));
+ appendExecCmdParameters(cmd, "htablename", seg.getStorageLocationIdentifier());
+ appendExecCmdParameters(cmd, "jobname", "Kylin_HFile_Generator_" + seg.getCubeInstance().getName() + "_Step");
+
+ createHFilesStep.setMapReduceParams(cmd.toString());
+ createHFilesStep.setMapReduceJobClass(CubeHFileJob.class);
+ createHFilesStep.setCounterSaveAs(",," + CubingJob.CUBE_SIZE_BYTES);
+
+ return createHFilesStep;
+ }
+
+ protected HadoopShellExecutable createBulkLoadStep(String jobId) {
+ HadoopShellExecutable bulkLoadStep = new HadoopShellExecutable();
+ bulkLoadStep.setName(ExecutableConstants.STEP_NAME_BULK_LOAD_HFILE);
+
+ StringBuilder cmd = new StringBuilder();
+ appendExecCmdParameters(cmd, "input", getHFilePath(jobId));
+ appendExecCmdParameters(cmd, "htablename", seg.getStorageLocationIdentifier());
+ appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
+
+ bulkLoadStep.setJobParams(cmd.toString());
+ bulkLoadStep.setJobClass(BulkLoadJob.class);
+
+ return bulkLoadStep;
+ }
+
+ protected GarbageCollectionStep createGarbageCollectionStep(List<String> oldHtables, String interimTable) {
+ GarbageCollectionStep result = new GarbageCollectionStep();
+ result.setName(ExecutableConstants.STEP_NAME_GARBAGE_COLLECTION);
+ result.setOldHTables(oldHtables);
+ result.setOldHiveTable(interimTable);
+ return result;
+ }
+
+ protected String getJobWorkingDir(String jobId) {
+ return getJobWorkingDir(config, jobId);
+ }
+
+ protected String getCuboidRootPath(String jobId) {
+ return getJobWorkingDir(jobId) + "/" + seg.getCubeInstance().getName() + "/cuboid/";
+ }
+
+ protected String getCuboidRootPath(CubeSegment seg) {
+ return getCuboidRootPath(seg.getLastBuildJobID());
+ }
+
+ protected void appendMapReduceParameters(StringBuilder buf, CubeSegment seg) {
+ try {
+ String jobConf = config.getHadoopJobConfFilePath(seg.getCubeDesc().getModel().getCapacity());
+ if (jobConf != null && jobConf.length() > 0) {
+ buf.append(" -conf ").append(jobConf);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected String getFactDistinctColumnsPath(String jobId) {
+ return getJobWorkingDir(jobId) + "/" + seg.getCubeInstance().getName() + "/fact_distinct_columns";
+ }
+
+
+ protected String getStatisticsPath(String jobId) {
+ return getJobWorkingDir(jobId) + "/" + seg.getCubeInstance().getName() + "/statistics";
+ }
+
+
+ protected String getHFilePath(String jobId) {
+ return getJobWorkingDir(jobId) + "/" + seg.getCubeInstance().getName() + "/hfile/";
+ }
+
+ protected String getRowkeyDistributionOutputPath(String jobId) {
+ return getJobWorkingDir(jobId) + "/" + seg.getCubeInstance().getName() + "/rowkey_stats";
+ }
+
+ // ============================================================================
+ // static methods also shared by IIJobBuilder
+ // ----------------------------------------------------------------------------
+
+ public static AbstractExecutable createFlatHiveTableStep(JobEngineConfig conf, IJoinedFlatTableDesc flatTableDesc, String jobId) {
+
+ final String dropTableHql = JoinedFlatTable.generateDropTableStatement(flatTableDesc, jobId);
+ final String createTableHql = JoinedFlatTable.generateCreateTableStatement(flatTableDesc, getJobWorkingDir(conf, jobId), jobId);
+ String insertDataHqls;
+ try {
+ insertDataHqls = JoinedFlatTable.generateInsertDataStatement(flatTableDesc, jobId, conf);
+ } catch (IOException e1) {
+ e1.printStackTrace();
+ throw new RuntimeException("Failed to generate insert data SQL for intermediate table.");
+ }
+
+ ShellExecutable step = new ShellExecutable();
+ StringBuffer buf = new StringBuffer();
+ buf.append("hive -e \"");
+ buf.append(dropTableHql + "\n");
+ buf.append(createTableHql + "\n");
+ buf.append(insertDataHqls + "\n");
+ buf.append("\"");
+
+ step.setCmd(buf.toString());
+ step.setName(ExecutableConstants.STEP_NAME_CREATE_FLAT_HIVE_TABLE);
+
+ return step;
+ }
+
+ public static String getJobWorkingDir(JobEngineConfig conf, String jobId) {
+ return conf.getHdfsWorkingDirectory() + "/" + "kylin-" + jobId;
+ }
+
+ public static StringBuilder appendExecCmdParameters(StringBuilder buf, String paraName, String paraValue) {
+ return buf.append(" -").append(paraName).append(" ").append(paraValue);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/mr/MRBuildEngine.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/mr/MRBuildEngine.java b/job/src/main/java/org/apache/kylin/engine/mr/MRBuildEngine.java
index 80b9009..3c482b4 100644
--- a/job/src/main/java/org/apache/kylin/engine/mr/MRBuildEngine.java
+++ b/job/src/main/java/org/apache/kylin/engine/mr/MRBuildEngine.java
@@ -18,14 +18,26 @@
package org.apache.kylin.engine.mr;
-import org.apache.kylin.engine.IBatchCubeBuilder;
+import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.engine.IBuildEngine;
+import org.apache.kylin.job.execution.DefaultChainedExecutable;
public class MRBuildEngine implements IBuildEngine {
@Override
- public IBatchCubeBuilder createBatchCubeBuilder() {
- return new MRCubeBuilder();
+ public Runnable createStreamingCubeBuilder(CubeSegment newSegment, String submitter) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public DefaultChainedExecutable createBatchBuildJob(CubeSegment newSegment, String submitter) {
+ return new BatchCubingJobBuilder(newSegment, submitter).build();
+ }
+
+ @Override
+ public DefaultChainedExecutable createBatchMergeJob(CubeSegment mergeSegment, String submitter) {
+ return new BatchMergeJobBuilder(mergeSegment, submitter).build();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/mr/MRCubeBuilder.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/mr/MRCubeBuilder.java b/job/src/main/java/org/apache/kylin/engine/mr/MRCubeBuilder.java
deleted file mode 100644
index 6bf0423..0000000
--- a/job/src/main/java/org/apache/kylin/engine/mr/MRCubeBuilder.java
+++ /dev/null
@@ -1,45 +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;
-
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.engine.IBatchCubeBuilder;
-import org.apache.kylin.job.execution.ChainedExecutable;
-
-public class MRCubeBuilder implements IBatchCubeBuilder {
-
- @Override
- public ChainedExecutable createBuildJob(CubeSegment newSegment) {
- // TODO Auto-generated method stub
- return null;
- }
-
- @Override
- public ChainedExecutable createMergeJob(CubeSegment mergeSegment) {
- // TODO Auto-generated method stub
- return null;
- }
-
- /** Build a new segment and merge it with existing segment into one step. Deprecated, only needed by Holistic Distinct Count. */
- public ChainedExecutable createBuildAndMergeJob(CubeSegment appendSegment, CubeSegment mergeSegment) {
- return null;
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/mr/MergeDictionaryStep.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/mr/MergeDictionaryStep.java b/job/src/main/java/org/apache/kylin/engine/mr/MergeDictionaryStep.java
new file mode 100644
index 0000000..0010174
--- /dev/null
+++ b/job/src/main/java/org/apache/kylin/engine/mr/MergeDictionaryStep.java
@@ -0,0 +1,197 @@
+/*
+ * 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;
+
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.StringUtils;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.cube.CubeInstance;
+import org.apache.kylin.cube.CubeBuilder;
+import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.model.CubeDesc;
+import org.apache.kylin.cube.model.DimensionDesc;
+import org.apache.kylin.dict.DictionaryInfo;
+import org.apache.kylin.dict.DictionaryManager;
+import org.apache.kylin.job.exception.ExecuteException;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.ExecutableContext;
+import org.apache.kylin.job.execution.ExecuteResult;
+import org.apache.kylin.metadata.model.TblColRef;
+
+import java.io.IOException;
+import java.util.*;
+
+public class MergeDictionaryStep extends AbstractExecutable {
+
+ private static final String CUBE_NAME = "cubeName";
+ private static final String SEGMENT_ID = "segmentId";
+ private static final String MERGING_SEGMENT_IDS = "mergingSegmentIds";
+
+ public MergeDictionaryStep() {
+ super();
+ }
+
+ @Override
+ protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException {
+ KylinConfig conf = context.getConfig();
+ final CubeManager mgr = CubeManager.getInstance(conf);
+ final CubeInstance cube = mgr.getCube(getCubeName());
+ final CubeSegment newSegment = cube.getSegmentById(getSegmentId());
+ final List<CubeSegment> mergingSegments = getMergingSegments(cube);
+
+ Collections.sort(mergingSegments);
+
+ try {
+ checkLookupSnapshotsMustIncremental(mergingSegments);
+
+ makeDictForNewSegment(conf, cube, newSegment, mergingSegments);
+ makeSnapshotForNewSegment(cube, newSegment, mergingSegments);
+
+ CubeBuilder cubeBuilder = new CubeBuilder(cube);
+ cubeBuilder.setToUpdateSegs(newSegment);
+ mgr.updateCube(cubeBuilder);
+ return new ExecuteResult(ExecuteResult.State.SUCCEED, "succeed");
+ } catch (IOException e) {
+ logger.error("fail to merge dictionary or lookup snapshots", e);
+ return new ExecuteResult(ExecuteResult.State.ERROR, e.getLocalizedMessage());
+ }
+ }
+
+ private List<CubeSegment> getMergingSegments(CubeInstance cube) {
+ List<String> mergingSegmentIds = getMergingSegmentIds();
+ List<CubeSegment> result = Lists.newArrayListWithCapacity(mergingSegmentIds.size());
+ for (String id : mergingSegmentIds) {
+ result.add(cube.getSegmentById(id));
+ }
+ return result;
+ }
+
+ private void checkLookupSnapshotsMustIncremental(List<CubeSegment> mergingSegments) {
+
+ // FIXME check each newer snapshot has only NEW rows but no MODIFIED rows
+ }
+
+ /**
+ * For the new segment, we need to create dictionaries for it, too. For
+ * those dictionaries on fact table, create it by merging underlying
+ * dictionaries For those dictionaries on lookup table, just copy it from
+ * any one of the merging segments, it's guaranteed to be consistent(checked
+ * in CubeSegmentValidator)
+ *
+ * @param cube
+ * @param newSeg
+ * @throws IOException
+ */
+ private void makeDictForNewSegment(KylinConfig conf, CubeInstance cube, CubeSegment newSeg, List<CubeSegment> mergingSegments) throws IOException {
+ HashSet<TblColRef> colsNeedMeringDict = new HashSet<TblColRef>();
+ HashSet<TblColRef> colsNeedCopyDict = new HashSet<TblColRef>();
+ DictionaryManager dictMgr = DictionaryManager.getInstance(conf);
+
+ CubeDesc cubeDesc = cube.getDescriptor();
+ for (DimensionDesc dim : cubeDesc.getDimensions()) {
+ for (TblColRef col : dim.getColumnRefs()) {
+ if (newSeg.getCubeDesc().getRowkey().isUseDictionary(col)) {
+ String dictTable = (String) dictMgr.decideSourceData(cubeDesc.getModel(), cubeDesc.getRowkey().getDictionary(col), col, null)[0];
+ if (cubeDesc.getFactTable().equalsIgnoreCase(dictTable)) {
+ colsNeedMeringDict.add(col);
+ } else {
+ colsNeedCopyDict.add(col);
+ }
+ }
+ }
+ }
+
+ for (TblColRef col : colsNeedMeringDict) {
+ logger.info("Merging fact table dictionary on : " + col);
+ List<DictionaryInfo> dictInfos = new ArrayList<DictionaryInfo>();
+ for (CubeSegment segment : mergingSegments) {
+ logger.info("Including fact table dictionary of segment : " + segment);
+ if (segment.getDictResPath(col) != null) {
+ DictionaryInfo dictInfo = dictMgr.getDictionaryInfo(segment.getDictResPath(col));
+ dictInfos.add(dictInfo);
+ }
+ }
+ mergeDictionaries(dictMgr, newSeg, dictInfos, col);
+ }
+
+ for (TblColRef col : colsNeedCopyDict) {
+ String path = mergingSegments.get(0).getDictResPath(col);
+ newSeg.putDictResPath(col, path);
+ }
+ }
+
+ private DictionaryInfo mergeDictionaries(DictionaryManager dictMgr, CubeSegment cubeSeg, List<DictionaryInfo> dicts, TblColRef col) throws IOException {
+ DictionaryInfo dictInfo = dictMgr.mergeDictionary(dicts);
+ if (dictInfo != null)
+ cubeSeg.putDictResPath(col, dictInfo.getResourcePath());
+
+ return dictInfo;
+ }
+
+ /**
+ * make snapshots for the new segment by copying from one of the underlying
+ * merging segments. it's guaranteed to be consistent(checked in
+ * CubeSegmentValidator)
+ *
+ * @param cube
+ * @param newSeg
+ */
+ private void makeSnapshotForNewSegment(CubeInstance cube, CubeSegment newSeg, List<CubeSegment> mergingSegments) {
+ CubeSegment lastSeg = mergingSegments.get(mergingSegments.size() - 1);
+ for (Map.Entry<String, String> entry : lastSeg.getSnapshots().entrySet()) {
+ newSeg.putSnapshotResPath(entry.getKey(), entry.getValue());
+ }
+ }
+
+ public void setCubeName(String cubeName) {
+ this.setParam(CUBE_NAME, cubeName);
+ }
+
+ private String getCubeName() {
+ return getParam(CUBE_NAME);
+ }
+
+ public void setSegmentId(String segmentId) {
+ this.setParam(SEGMENT_ID, segmentId);
+ }
+
+ private String getSegmentId() {
+ return getParam(SEGMENT_ID);
+ }
+
+ public void setMergingSegmentIds(List<String> ids) {
+ setParam(MERGING_SEGMENT_IDS, StringUtils.join(ids, ","));
+ }
+
+ private List<String> getMergingSegmentIds() {
+ final String ids = getParam(MERGING_SEGMENT_IDS);
+ if (ids != null) {
+ final String[] splitted = StringUtils.split(ids, ",");
+ ArrayList<String> result = Lists.newArrayListWithExpectedSize(splitted.length);
+ for (String id: splitted) {
+ result.add(id);
+ }
+ return result;
+ } else {
+ return Collections.emptyList();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/mr/UpdateCubeInfoAfterBuildStep.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/mr/UpdateCubeInfoAfterBuildStep.java b/job/src/main/java/org/apache/kylin/engine/mr/UpdateCubeInfoAfterBuildStep.java
new file mode 100644
index 0000000..8361f7c
--- /dev/null
+++ b/job/src/main/java/org/apache/kylin/engine/mr/UpdateCubeInfoAfterBuildStep.java
@@ -0,0 +1,100 @@
+/*
+ * 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;
+
+import java.io.IOException;
+
+import org.apache.kylin.cube.CubeBuilder;
+import org.apache.kylin.cube.CubeInstance;
+import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.job.exception.ExecuteException;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.ExecutableContext;
+import org.apache.kylin.job.execution.ExecuteResult;
+
+/**
+ */
+public class UpdateCubeInfoAfterBuildStep extends AbstractExecutable {
+
+ private static final String SEGMENT_ID = "segmentId";
+ private static final String CUBE_NAME = "cubeName";
+ private static final String CUBING_JOB_ID = "cubingJobId";
+
+ public UpdateCubeInfoAfterBuildStep() {
+ super();
+ }
+
+ public void setCubeName(String cubeName) {
+ this.setParam(CUBE_NAME, cubeName);
+ }
+
+ private String getCubeName() {
+ return getParam(CUBE_NAME);
+ }
+
+ public void setSegmentId(String segmentId) {
+ this.setParam(SEGMENT_ID, segmentId);
+ }
+
+ private String getSegmentId() {
+ return getParam(SEGMENT_ID);
+ }
+
+ public void setCubingJobId(String id) {
+ setParam(CUBING_JOB_ID, id);
+ }
+
+ private String getCubingJobId() {
+ return getParam(CUBING_JOB_ID);
+ }
+
+ @Override
+ protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException {
+ final CubeManager cubeManager = CubeManager.getInstance(context.getConfig());
+ final CubeInstance cube = cubeManager.getCube(getCubeName());
+ final CubeSegment segment = cube.getSegmentById(getSegmentId());
+
+ CubingJob cubingJob = (CubingJob) executableManager.getJob(getCubingJobId());
+ long sourceCount = cubingJob.findSourceRecordCount();
+ long sourceSizeBytes = cubingJob.findSourceSizeBytes();
+ long cubeSizeBytes = cubingJob.findCubeSizeBytes();
+ boolean segmentReady = cubeSizeBytes > 0; // for build+merge scenario, convert HFile not happen yet, so cube size is 0
+
+ segment.setLastBuildJobID(getCubingJobId());
+ segment.setLastBuildTime(System.currentTimeMillis());
+ segment.setSizeKB(cubeSizeBytes / 1024);
+ segment.setInputRecords(sourceCount);
+ segment.setInputRecordsSize(sourceSizeBytes);
+
+ try {
+ if (segmentReady) {
+ cubeManager.promoteNewlyBuiltSegments(cube, segment);
+ } else {
+ CubeBuilder cubeBuilder = new CubeBuilder(cube);
+ cubeBuilder.setToUpdateSegs(segment);
+ cubeManager.updateCube(cubeBuilder);
+ }
+ return new ExecuteResult(ExecuteResult.State.SUCCEED, "succeed");
+ } catch (IOException e) {
+ logger.error("fail to update cube after build", e);
+ return new ExecuteResult(ExecuteResult.State.ERROR, e.getLocalizedMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/engine/mr/UpdateCubeInfoAfterMergeStep.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/engine/mr/UpdateCubeInfoAfterMergeStep.java b/job/src/main/java/org/apache/kylin/engine/mr/UpdateCubeInfoAfterMergeStep.java
new file mode 100644
index 0000000..7defec4
--- /dev/null
+++ b/job/src/main/java/org/apache/kylin/engine/mr/UpdateCubeInfoAfterMergeStep.java
@@ -0,0 +1,135 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.commons.lang.StringUtils;
+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.job.exception.ExecuteException;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.ExecutableContext;
+import org.apache.kylin.job.execution.ExecuteResult;
+
+import com.google.common.collect.Lists;
+
+/**
+ */
+public class UpdateCubeInfoAfterMergeStep extends AbstractExecutable {
+
+ private static final String CUBE_NAME = "cubeName";
+ private static final String SEGMENT_ID = "segmentId";
+ private static final String MERGING_SEGMENT_IDS = "mergingSegmentIds";
+ private static final String CUBING_JOB_ID = "cubingJobId";
+
+ private final CubeManager cubeManager = CubeManager.getInstance(KylinConfig.getInstanceFromEnv());
+
+ public UpdateCubeInfoAfterMergeStep() {
+ super();
+ }
+
+ @Override
+ protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException {
+ final CubeInstance cube = cubeManager.getCube(getCubeName());
+
+ CubeSegment mergedSegment = cube.getSegmentById(getSegmentId());
+ if (mergedSegment == null) {
+ return new ExecuteResult(ExecuteResult.State.FAILED, "there is no segment with id:" + getSegmentId());
+ }
+
+ CubingJob cubingJob = (CubingJob) executableManager.getJob(getCubingJobId());
+ long cubeSizeBytes = cubingJob.findCubeSizeBytes();
+
+ // collect source statistics
+ List<String> mergingSegmentIds = getMergingSegmentIds();
+ if (mergingSegmentIds.isEmpty()) {
+ return new ExecuteResult(ExecuteResult.State.FAILED, "there are no merging segments");
+ }
+ long sourceCount = 0L;
+ long sourceSize = 0L;
+ for (String id : mergingSegmentIds) {
+ CubeSegment segment = cube.getSegmentById(id);
+ sourceCount += segment.getInputRecords();
+ sourceSize += segment.getInputRecordsSize();
+ }
+
+ // update segment info
+ mergedSegment.setSizeKB(cubeSizeBytes / 1024);
+ mergedSegment.setInputRecords(sourceCount);
+ mergedSegment.setInputRecordsSize(sourceSize);
+ mergedSegment.setLastBuildJobID(getCubingJobId());
+ mergedSegment.setLastBuildTime(System.currentTimeMillis());
+
+ try {
+ cubeManager.promoteNewlyBuiltSegments(cube, mergedSegment);
+ return new ExecuteResult(ExecuteResult.State.SUCCEED);
+ } catch (IOException e) {
+ logger.error("fail to update cube after merge", e);
+ return new ExecuteResult(ExecuteResult.State.ERROR, e.getLocalizedMessage());
+ }
+ }
+
+ public void setSegmentId(String segmentId) {
+ this.setParam(SEGMENT_ID, segmentId);
+ }
+
+ private String getSegmentId() {
+ return getParam(SEGMENT_ID);
+ }
+
+ public void setCubeName(String cubeName) {
+ this.setParam(CUBE_NAME, cubeName);
+ }
+
+ private String getCubeName() {
+ return getParam(CUBE_NAME);
+ }
+
+ public void setMergingSegmentIds(List<String> ids) {
+ setParam(MERGING_SEGMENT_IDS, StringUtils.join(ids, ","));
+ }
+
+ private List<String> getMergingSegmentIds() {
+ final String ids = getParam(MERGING_SEGMENT_IDS);
+ if (ids != null) {
+ final String[] splitted = StringUtils.split(ids, ",");
+ ArrayList<String> result = Lists.newArrayListWithExpectedSize(splitted.length);
+ for (String id : splitted) {
+ result.add(id);
+ }
+ return result;
+ } else {
+ return Collections.emptyList();
+ }
+ }
+
+ 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/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/job/AbstractJobBuilder.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/job/AbstractJobBuilder.java b/job/src/main/java/org/apache/kylin/job/AbstractJobBuilder.java
index d7fe00e..48766b7 100644
--- a/job/src/main/java/org/apache/kylin/job/AbstractJobBuilder.java
+++ b/job/src/main/java/org/apache/kylin/job/AbstractJobBuilder.java
@@ -58,32 +58,6 @@ public abstract class AbstractJobBuilder {
return getJobWorkingDir(jobUUID) + "/" + intermediateTableDesc.getTableName(jobUUID);
}
- protected AbstractExecutable createIntermediateHiveTableStep(IJoinedFlatTableDesc intermediateTableDesc, String jobId) {
-
- final String dropTableHql = JoinedFlatTable.generateDropTableStatement(intermediateTableDesc, jobId);
- final String createTableHql = JoinedFlatTable.generateCreateTableStatement(intermediateTableDesc, getJobWorkingDir(jobId), jobId);
- String insertDataHqls;
- try {
- insertDataHqls = JoinedFlatTable.generateInsertDataStatement(intermediateTableDesc, jobId, this.engineConfig);
- } catch (IOException e1) {
- e1.printStackTrace();
- throw new RuntimeException("Failed to generate insert data SQL for intermediate table.");
- }
-
- ShellExecutable step = new ShellExecutable();
- StringBuffer buf = new StringBuffer();
- buf.append("hive -e \"");
- buf.append(dropTableHql + "\n");
- buf.append(createTableHql + "\n");
- buf.append(insertDataHqls + "\n");
- buf.append("\"");
-
- step.setCmd(buf.toString());
- step.setName(ExecutableConstants.STEP_NAME_CREATE_FLAT_HIVE_TABLE);
-
- return step;
- }
-
protected String getJobWorkingDir(String uuid) {
return engineConfig.getHdfsWorkingDirectory() + "/" + JOB_WORKING_DIR_PREFIX + uuid;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/87c32b0f/job/src/main/java/org/apache/kylin/job/CubeMetadataUpgrade.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/job/CubeMetadataUpgrade.java b/job/src/main/java/org/apache/kylin/job/CubeMetadataUpgrade.java
index f45b9ef..b00e10e 100644
--- a/job/src/main/java/org/apache/kylin/job/CubeMetadataUpgrade.java
+++ b/job/src/main/java/org/apache/kylin/job/CubeMetadataUpgrade.java
@@ -44,13 +44,13 @@ import org.apache.kylin.cube.model.v1.CubeInstance;
import org.apache.kylin.cube.model.v1.CubeSegment;
import org.apache.kylin.cube.model.v1.CubeSegmentStatusEnum;
import org.apache.kylin.cube.model.v1.CubeStatusEnum;
+import org.apache.kylin.engine.mr.CubingJob;
import org.apache.kylin.job.common.HadoopShellExecutable;
import org.apache.kylin.job.common.MapReduceExecutable;
import org.apache.kylin.job.common.ShellExecutable;
import org.apache.kylin.job.constant.ExecutableConstants;
import org.apache.kylin.job.constant.JobStatusEnum;
import org.apache.kylin.job.constant.JobStepStatusEnum;
-import org.apache.kylin.job.cube.CubingJob;
import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.ExecutableState;
import org.apache.kylin.job.hadoop.cube.BaseCuboidJob;