KYLIN-1116 Use local dictionary for InvertedIndex batch building Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/82bfa924 Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/82bfa924 Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/82bfa924
Branch: refs/heads/KYLIN-1116 Commit: 82bfa924c48db87297be6a7720357b500d48511e Parents: f8590d2 Author: shaofengshi <shaofeng...@apache.org> Authored: Fri Nov 6 16:24:29 2015 +0800 Committer: shaofengshi <shaofeng...@apache.org> Committed: Fri Nov 6 16:24:29 2015 +0800 ---------------------------------------------------------------------- .../apache/kylin/job/BuildIIWithStreamTest.java | 15 +- .../kylin/job/hadoop/invertedindex/IITest.java | 11 +- .../kylin/common/util/StreamingBatch.java | 59 +++++ .../kylin/common/util/StreamingMessage.java | 43 ++++ .../realization/IRealizationSegment.java | 1 - .../kylin/engine/mr/JobBuilderSupport.java | 8 +- .../engine/mr/common/AbstractHadoopJob.java | 3 - .../mr/invertedindex/BatchIIJobBuilder.java | 61 +----- .../CreateInvertedIndexDictionaryJob.java | 70 ------ .../IIDistinctColumnsCombiner.java | 58 ----- .../mr/invertedindex/IIDistinctColumnsJob.java | 138 ------------ .../invertedindex/IIDistinctColumnsMapper.java | 75 ------- .../invertedindex/IIDistinctColumnsReducer.java | 77 ------- .../engine/mr/invertedindex/IIJobBuilder.java | 219 ------------------- .../mr/invertedindex/InvertedIndexJob.java | 10 +- .../mr/invertedindex/InvertedIndexMapper.java | 32 +-- .../mr/invertedindex/InvertedIndexReducer.java | 89 +++++--- .../UpdateIIInfoAfterBuildStep.java | 84 +++++++ .../UpdateInvertedIndexInfoAfterBuildStep.java | 93 -------- .../kylin/engine/streaming/IStreamingInput.java | 2 + .../streaming/OneOffStreamingBuilder.java | 1 + .../kylin/engine/streaming/StreamingBatch.java | 61 ------ .../engine/streaming/StreamingBatchBuilder.java | 1 + .../engine/streaming/StreamingMessage.java | 43 ---- .../streaming/cube/StreamingCubeBuilder.java | 4 +- .../streaming/invertedindex/SliceBuilder.java | 81 ------- .../invertedindex/test_kylin_ii_inner_join.json | 45 +--- .../invertedindex/test_kylin_ii_left_join.json | 45 +--- .../invertedindex/test_streaming_table_ii.json | 22 +- .../apache/kylin/invertedindex/IIManager.java | 50 +---- .../apache/kylin/invertedindex/IISegment.java | 51 +---- .../kylin/invertedindex/index/SliceBuilder.java | 77 +++++++ .../invertedindex/IIInstanceTest.java | 5 - .../apache/kylin/source/hive/HiveMRInput.java | 3 + .../kylin/source/kafka/KafkaStreamingInput.java | 4 +- .../kylin/source/kafka/StreamingParser.java | 2 +- .../source/kafka/StringStreamingParser.java | 2 +- .../source/kafka/TimedJsonStreamParser.java | 2 +- .../kafka/diagnose/KafkaInputAnalyzer.java | 2 +- .../kylin/source/kafka/util/KafkaUtils.java | 2 +- .../storage/hbase/ii/IICreateHFileMapper.java | 19 +- .../storage/hbase/ii/IICreateHTableJob.java | 31 ++- .../storage/hbase/steps/HBaseMROutput.java | 2 +- .../kylin/storage/hbase/steps/HBaseMRSteps.java | 18 +- .../hbase/util/DeployCoprocessorCLI.java | 3 +- .../storage/hbase/util/StorageCleanupJob.java | 2 - .../hbase/common/TsConditionEraserTest.java | 6 +- .../endpoint/EndpointAggregationTest.java | 27 ++- .../endpoint/TableRecordInfoTest.java | 5 + .../endpoint/TsConditionExtractorTest.java | 5 + 50 files changed, 469 insertions(+), 1300 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java ---------------------------------------------------------------------- diff --git a/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java b/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java index b64a7c5..2d40b09 100644 --- a/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java +++ b/assembly/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java @@ -51,8 +51,8 @@ import org.apache.kylin.common.util.ClassUtil; import org.apache.kylin.common.util.DateFormat; import org.apache.kylin.common.util.Pair; import org.apache.kylin.engine.mr.JobBuilderSupport; -import org.apache.kylin.engine.streaming.StreamingBatch; -import org.apache.kylin.engine.streaming.StreamingMessage; +import org.apache.kylin.common.util.StreamingBatch; +import org.apache.kylin.common.util.StreamingMessage; import org.apache.kylin.invertedindex.IIInstance; import org.apache.kylin.invertedindex.IIManager; import org.apache.kylin.invertedindex.IISegment; @@ -61,7 +61,7 @@ import org.apache.kylin.invertedindex.model.IIDesc; import org.apache.kylin.invertedindex.model.IIJoinedFlatTableDesc; import org.apache.kylin.invertedindex.model.IIKeyValueCodec; import org.apache.kylin.invertedindex.model.IIRow; -import org.apache.kylin.engine.streaming.invertedindex.SliceBuilder; +import org.apache.kylin.invertedindex.index.SliceBuilder; import org.apache.kylin.job.common.ShellExecutable; import org.apache.kylin.job.constant.ExecutableConstants; import org.apache.kylin.job.engine.JobEngineConfig; @@ -196,14 +196,13 @@ public class BuildIIWithStreamTest { int count = sorted.size(); ArrayList<StreamingMessage> messages = Lists.newArrayList(); for (String[] row : sorted) { - if (messages.size() < iiDesc.getSliceSize()) { - messages.add(parse(row)); - } else { + messages.add((parse(row))); + if (messages.size() >= iiDesc.getSliceSize()) { build(sliceBuilder, new StreamingBatch(messages, Pair.newPair(System.currentTimeMillis(), System.currentTimeMillis())), htable); - messages = Lists.newArrayList(); - messages.add((parse(row))); + messages.clear(); } } + if (!messages.isEmpty()) { build(sliceBuilder, new StreamingBatch(messages, Pair.newPair(System.currentTimeMillis(), System.currentTimeMillis())), htable); } http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/assembly/src/test/java/org/apache/kylin/job/hadoop/invertedindex/IITest.java ---------------------------------------------------------------------- diff --git a/assembly/src/test/java/org/apache/kylin/job/hadoop/invertedindex/IITest.java b/assembly/src/test/java/org/apache/kylin/job/hadoop/invertedindex/IITest.java index 200156a..a393ce3 100644 --- a/assembly/src/test/java/org/apache/kylin/job/hadoop/invertedindex/IITest.java +++ b/assembly/src/test/java/org/apache/kylin/job/hadoop/invertedindex/IITest.java @@ -23,10 +23,11 @@ import org.apache.hadoop.hbase.regionserver.RegionScanner; import org.apache.kylin.common.util.FIFOIterable; import org.apache.kylin.common.util.LocalFileMetadataTestCase; import org.apache.kylin.common.util.Pair; -import org.apache.kylin.engine.streaming.StreamingBatch; -import org.apache.kylin.engine.streaming.StreamingMessage; +import org.apache.kylin.common.util.StreamingBatch; +import org.apache.kylin.common.util.StreamingMessage; import org.apache.kylin.invertedindex.IIInstance; import org.apache.kylin.invertedindex.IIManager; +import org.apache.kylin.invertedindex.IISegment; import org.apache.kylin.invertedindex.index.Slice; import org.apache.kylin.invertedindex.index.TableRecordInfo; import org.apache.kylin.invertedindex.index.TableRecordInfoDigest; @@ -35,7 +36,7 @@ import org.apache.kylin.invertedindex.model.IIKeyValueCodec; import org.apache.kylin.invertedindex.model.IIKeyValueCodecWithState; import org.apache.kylin.invertedindex.model.IIRow; import org.apache.kylin.invertedindex.model.KeyValueCodec; -import org.apache.kylin.engine.streaming.invertedindex.SliceBuilder; +import org.apache.kylin.invertedindex.index.SliceBuilder; import org.apache.kylin.metadata.filter.ColumnTupleFilter; import org.apache.kylin.metadata.filter.CompareTupleFilter; import org.apache.kylin.metadata.filter.ConstantTupleFilter; @@ -146,6 +147,10 @@ public class IITest extends LocalFileMetadataTestCase { @Test public void IIEndpointTest() { TableRecordInfo info = new TableRecordInfo(ii.getDescriptor()); + if (ii.getFirstSegment() == null) { + IISegment segment = IIManager.getInstance(getTestConfig()).buildSegment(ii, 0, System.currentTimeMillis()); + ii.getSegments().add(segment); + } CoprocessorRowType type = CoprocessorRowType.fromTableRecordInfo(info, ii.getFirstSegment().getColumns()); CoprocessorProjector projector = CoprocessorProjector.makeForEndpoint(info, Collections.singletonList(ii.getDescriptor().findColumnRef("default.test_kylin_fact", "lstg_format_name"))); http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/core-common/src/main/java/org/apache/kylin/common/util/StreamingBatch.java ---------------------------------------------------------------------- diff --git a/core-common/src/main/java/org/apache/kylin/common/util/StreamingBatch.java b/core-common/src/main/java/org/apache/kylin/common/util/StreamingBatch.java new file mode 100644 index 0000000..ae5b72f --- /dev/null +++ b/core-common/src/main/java/org/apache/kylin/common/util/StreamingBatch.java @@ -0,0 +1,59 @@ +/* + * + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * + * contributor license agreements. See the NOTICE file distributed with + * + * this work for additional information regarding copyright ownership. + * + * The ASF licenses this file to You under the Apache License, Version 2.0 + * + * (the "License"); you may not use this file except in compliance with + * + * the License. You may obtain a copy of the License at + * + * + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * + * + * Unless required by applicable law or agreed to in writing, software + * + * distributed under the License is distributed on an "AS IS" BASIS, + * + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * + * See the License for the specific language governing permissions and + * + * limitations under the License. + * + * / + */ +package org.apache.kylin.common.util; + +import java.util.List; + +/** + */ +public final class StreamingBatch { + + private final List<StreamingMessage> messages; + + private final Pair<Long, Long> timeRange; + + public StreamingBatch(List<StreamingMessage> messages, Pair<Long, Long> timeRange) { + this.messages = messages; + this.timeRange = timeRange; + } + + public List<StreamingMessage> getMessages() { + return messages; + } + + public Pair<Long, Long> getTimeRange() { + return timeRange; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessage.java ---------------------------------------------------------------------- diff --git a/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessage.java b/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessage.java new file mode 100644 index 0000000..2c150ff --- /dev/null +++ b/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessage.java @@ -0,0 +1,43 @@ +package org.apache.kylin.common.util; + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +/** + */ +public class StreamingMessage { + + private final List<String> data; + + private long offset; + + private long timestamp; + + private Map<String, Object> params; + + public static final StreamingMessage EOF = new StreamingMessage(Collections.<String> emptyList(), 0L, 0L, Collections.<String, Object> emptyMap()); + + public StreamingMessage(List<String> data, long offset, long timestamp, Map<String, Object> params) { + this.data = data; + this.offset = offset; + this.timestamp = timestamp; + this.params = params; + } + + public final List<String> getData() { + return data; + } + + public final long getOffset() { + return offset; + } + + public final long getTimestamp() { + return timestamp; + } + + public Map<String, Object> getParams() { + return params; + } +} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/core-metadata/src/main/java/org/apache/kylin/metadata/realization/IRealizationSegment.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/realization/IRealizationSegment.java b/core-metadata/src/main/java/org/apache/kylin/metadata/realization/IRealizationSegment.java index afab86b..3ac82a0 100644 --- a/core-metadata/src/main/java/org/apache/kylin/metadata/realization/IRealizationSegment.java +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/realization/IRealizationSegment.java @@ -4,7 +4,6 @@ import org.apache.kylin.metadata.model.IBuildable; import org.apache.kylin.metadata.model.IJoinedFlatTableDesc; /** - * Created by shaoshi on 10/30/15. */ public interface IRealizationSegment extends IBuildable { http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java index 8c770f9..e3b07d8 100644 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java +++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java @@ -25,13 +25,12 @@ import org.apache.kylin.common.KylinConfig; import org.apache.kylin.cube.CubeSegment; import org.apache.kylin.engine.mr.common.HadoopShellExecutable; import org.apache.kylin.engine.mr.common.MapReduceExecutable; -import org.apache.kylin.engine.mr.invertedindex.UpdateInvertedIndexInfoAfterBuildStep; +import org.apache.kylin.engine.mr.invertedindex.UpdateIIInfoAfterBuildStep; import org.apache.kylin.engine.mr.steps.CreateDictionaryJob; import org.apache.kylin.engine.mr.steps.FactDistinctColumnsJob; import org.apache.kylin.engine.mr.steps.MergeDictionaryStep; import org.apache.kylin.engine.mr.steps.UpdateCubeInfoAfterBuildStep; import org.apache.kylin.engine.mr.steps.UpdateCubeInfoAfterMergeStep; -import org.apache.kylin.invertedindex.IISegment; import org.apache.kylin.job.constant.ExecutableConstants; import org.apache.kylin.job.engine.JobEngineConfig; @@ -125,11 +124,10 @@ public class JobBuilderSupport { - public UpdateInvertedIndexInfoAfterBuildStep createUpdateInvertedIndexInfoAfterBuildStep(String jobId) { - final UpdateInvertedIndexInfoAfterBuildStep updateIIInfoStep = new UpdateInvertedIndexInfoAfterBuildStep(); + public UpdateIIInfoAfterBuildStep createUpdateIIInfoAfterBuildStep(String jobId) { + final UpdateIIInfoAfterBuildStep updateIIInfoStep = new UpdateIIInfoAfterBuildStep(); updateIIInfoStep.setName(ExecutableConstants.STEP_NAME_UPDATE_II_INFO); updateIIInfoStep.setInvertedIndexName(seg.getRealization().getName()); - updateIIInfoStep.setSegmentId(seg.getUuid()); updateIIInfoStep.setJobId(jobId); return updateIIInfoStep; } http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/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 8782bbe..366a730 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 @@ -337,9 +337,6 @@ public abstract class AbstractHadoopJob extends Configured implements Tool { TableDesc table = metaMgr.getTableDesc(tableName); dumpList.add(table.getResourcePath()); } - for (IISegment segment : ii.getSegments()) { - dumpList.addAll(segment.getDictionaryPaths()); - } attachKylinPropsAndMetadata(dumpList, conf); } http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/BatchIIJobBuilder.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/BatchIIJobBuilder.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/BatchIIJobBuilder.java index 97e27d0..e7501b8 100644 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/BatchIIJobBuilder.java +++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/BatchIIJobBuilder.java @@ -18,30 +18,16 @@ package org.apache.kylin.engine.mr.invertedindex; -import org.apache.kylin.cube.CubeSegment; -import org.apache.kylin.engine.mr.CubingJob; import org.apache.kylin.engine.mr.IMRInput.IMRBatchCubingInputSide; import org.apache.kylin.engine.mr.IMROutput; -import org.apache.kylin.engine.mr.IMROutput.IMRBatchCubingOutputSide; import org.apache.kylin.engine.mr.JobBuilderSupport; import org.apache.kylin.engine.mr.MRUtil; -import org.apache.kylin.engine.mr.common.HadoopShellExecutable; import org.apache.kylin.engine.mr.common.MapReduceExecutable; -import org.apache.kylin.engine.mr.steps.BaseCuboidJob; -import org.apache.kylin.engine.mr.steps.NDCuboidJob; import org.apache.kylin.invertedindex.IISegment; -import org.apache.kylin.invertedindex.model.IIJoinedFlatTableDesc; 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.metadata.model.DataModelDesc; -import org.apache.kylin.metadata.realization.IRealization; -import org.apache.kylin.metadata.realization.IRealizationSegment; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; - public class BatchIIJobBuilder extends JobBuilderSupport { private static final Logger logger = LoggerFactory.getLogger(BatchIIJobBuilder.class); @@ -64,54 +50,20 @@ public class BatchIIJobBuilder extends JobBuilderSupport { final String iiRootPath = getRealizationRootPath(jobId) + "/"; // Phase 1: Create Flat Table inputSide.addStepPhase1_CreateFlatTable(result); - - final String intermediateTableIdentity = seg.getJoinedFlatTableDesc().getTableName(); - // Phase 2: Build Dictionary - result.addTask(createIIFactDistinctColumnsStep(seg, intermediateTableIdentity, getFactDistinctColumnsPath(jobId))); - result.addTask(createIIBuildDictionaryStep(seg, getFactDistinctColumnsPath(jobId))); - - // Phase 3: Build Cube - result.addTask(createInvertedIndexStep((IISegment)seg, intermediateTableIdentity, iiRootPath)); + + // Phase 2: Build Inverted Index + result.addTask(createInvertedIndexStep((IISegment)seg, iiRootPath)); outputSide.addStepPhase3_BuildII(result, iiRootPath); - // Phase 4: Update Metadata & Cleanup - result.addTask(createUpdateInvertedIndexInfoAfterBuildStep(jobId)); + // Phase 3: Update Metadata & Cleanup + result.addTask(createUpdateIIInfoAfterBuildStep(jobId)); inputSide.addStepPhase4_Cleanup(result); outputSide.addStepPhase4_Cleanup(result); return result; } - private MapReduceExecutable createIIFactDistinctColumnsStep(IRealizationSegment seg, String factTableName, String output) { - MapReduceExecutable result = new MapReduceExecutable(); - result.setName(ExecutableConstants.STEP_NAME_FACT_DISTINCT_COLUMNS); - result.setMapReduceJobClass(IIDistinctColumnsJob.class); - StringBuilder cmd = new StringBuilder(); - appendMapReduceParameters(cmd, seg.getRealization().getDataModelDesc()); - appendExecCmdParameters(cmd, "tablename", factTableName); - appendExecCmdParameters(cmd, "iiname", seg.getRealization().getName()); - appendExecCmdParameters(cmd, "output", output); - appendExecCmdParameters(cmd, "jobname", "Kylin_Fact_Distinct_Columns_" + seg.getRealization().getName() + "_Step"); - - result.setMapReduceParams(cmd.toString()); - return result; - } - - private HadoopShellExecutable createIIBuildDictionaryStep(IRealizationSegment seg, String factDistinctColumnsPath) { - // base cuboid job - HadoopShellExecutable buildDictionaryStep = new HadoopShellExecutable(); - buildDictionaryStep.setName(ExecutableConstants.STEP_NAME_BUILD_DICTIONARY); - StringBuilder cmd = new StringBuilder(); - appendExecCmdParameters(cmd, "iiname", seg.getRealization().getName()); - appendExecCmdParameters(cmd, "input", factDistinctColumnsPath); - - buildDictionaryStep.setJobParams(cmd.toString()); - buildDictionaryStep.setJobClass(CreateInvertedIndexDictionaryJob.class); - return buildDictionaryStep; - } - - private MapReduceExecutable createInvertedIndexStep(IISegment seg, String intermediateHiveTable, String iiOutputTempPath) { - // base cuboid job + private MapReduceExecutable createInvertedIndexStep(IISegment seg, String iiOutputTempPath) { MapReduceExecutable buildIIStep = new MapReduceExecutable(); StringBuilder cmd = new StringBuilder(); @@ -120,7 +72,6 @@ public class BatchIIJobBuilder extends JobBuilderSupport { buildIIStep.setName(ExecutableConstants.STEP_NAME_BUILD_II); appendExecCmdParameters(cmd, "iiname", seg.getRealization().getName()); - appendExecCmdParameters(cmd, "tablename", intermediateHiveTable); appendExecCmdParameters(cmd, "output", iiOutputTempPath); appendExecCmdParameters(cmd, "jobname", ExecutableConstants.STEP_NAME_BUILD_II); http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/CreateInvertedIndexDictionaryJob.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/CreateInvertedIndexDictionaryJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/CreateInvertedIndexDictionaryJob.java deleted file mode 100644 index 39d74b4..0000000 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/CreateInvertedIndexDictionaryJob.java +++ /dev/null @@ -1,70 +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.invertedindex; - -import org.apache.commons.cli.Options; -import org.apache.hadoop.util.ToolRunner; -import org.apache.kylin.common.KylinConfig; -import org.apache.kylin.dict.DistinctColumnValuesProvider; -import org.apache.kylin.engine.mr.DFSFileTable; -import org.apache.kylin.engine.mr.common.AbstractHadoopJob; -import org.apache.kylin.invertedindex.IIInstance; -import org.apache.kylin.invertedindex.IIManager; -import org.apache.kylin.metadata.model.TblColRef; -import org.apache.kylin.source.ReadableTable; - -/** - */ -public class CreateInvertedIndexDictionaryJob extends AbstractHadoopJob { - - @Override - public int run(String[] args) throws Exception { - Options options = new Options(); - - try { - options.addOption(OPTION_II_NAME); - options.addOption(OPTION_INPUT_PATH); - parseOptions(options, args); - - final String iiname = getOptionValue(OPTION_II_NAME); - final String factColumnsInputPath = getOptionValue(OPTION_INPUT_PATH); - final KylinConfig config = KylinConfig.getInstanceFromEnv(); - - IIManager mgr = IIManager.getInstance(config); - IIInstance ii = mgr.getII(iiname); - - mgr.buildInvertedIndexDictionary(ii.getFirstSegment(), new DistinctColumnValuesProvider() { - @Override - public ReadableTable getDistinctValuesFor(TblColRef col) { - return new DFSFileTable(factColumnsInputPath + "/" + col.getName(), -1); - } - }); - return 0; - } catch (Exception e) { - printUsage(options); - throw e; - } - } - - public static void main(String[] args) throws Exception { - int exitCode = ToolRunner.run(new CreateInvertedIndexDictionaryJob(), args); - System.exit(exitCode); - } - -} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsCombiner.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsCombiner.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsCombiner.java deleted file mode 100644 index 651ad63..0000000 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsCombiner.java +++ /dev/null @@ -1,58 +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.invertedindex; - -import java.io.IOException; -import java.util.HashSet; - -import org.apache.hadoop.io.ShortWritable; -import org.apache.hadoop.io.Text; -import org.apache.kylin.common.util.ByteArray; -import org.apache.kylin.common.util.Bytes; -import org.apache.kylin.engine.mr.KylinReducer; - -/** - * @author yangli9 - */ -public class IIDistinctColumnsCombiner extends KylinReducer<ShortWritable, Text, ShortWritable, Text> { - - private Text outputValue = new Text(); - - @Override - protected void setup(Context context) throws IOException { - super.bindCurrentConfiguration(context.getConfiguration()); - - } - - @Override - public void reduce(ShortWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException { - - HashSet<ByteArray> set = new HashSet<ByteArray>(); - for (Text textValue : values) { - ByteArray value = new ByteArray(Bytes.copy(textValue.getBytes(), 0, textValue.getLength())); - set.add(value); - } - - for (ByteArray value : set) { - outputValue.set(value.array(), value.offset(), value.length()); - context.write(key, outputValue); - } - } - -} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsJob.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsJob.java deleted file mode 100644 index fe968b1..0000000 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsJob.java +++ /dev/null @@ -1,138 +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.invertedindex; - -import java.io.IOException; - -import org.apache.commons.cli.Options; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.io.NullWritable; -import org.apache.hadoop.io.ShortWritable; -import org.apache.hadoop.io.Text; -import org.apache.hadoop.mapreduce.Job; -import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; -import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; -import org.apache.hadoop.util.ToolRunner; -import org.apache.kylin.common.KylinConfig; -import org.apache.kylin.engine.mr.HadoopUtil; -import org.apache.kylin.engine.mr.IMRInput; -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.invertedindex.IIInstance; -import org.apache.kylin.invertedindex.IIManager; -import org.apache.kylin.invertedindex.IISegment; -import org.apache.kylin.invertedindex.model.IIJoinedFlatTableDesc; -import org.apache.kylin.metadata.model.IntermediateColumnDesc; -import org.apache.kylin.metadata.model.SegmentStatusEnum; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * @author yangli9 - */ -public class IIDistinctColumnsJob extends AbstractHadoopJob { - protected static final Logger logger = LoggerFactory.getLogger(IIDistinctColumnsJob.class); - - @Override - public int run(String[] args) throws Exception { - Options options = new Options(); - - try { - options.addOption(OPTION_JOB_NAME); - options.addOption(OPTION_TABLE_NAME); - options.addOption(OPTION_II_NAME); - options.addOption(OPTION_OUTPUT_PATH); - parseOptions(options, args); - - job = Job.getInstance(getConf(), getOptionValue(OPTION_JOB_NAME)); - String tableName = getOptionValue(OPTION_TABLE_NAME).toUpperCase(); - String iiName = getOptionValue(OPTION_II_NAME); - Path output = new Path(getOptionValue(OPTION_OUTPUT_PATH)); - - // ---------------------------------------------------------------------------- - - logger.info("Starting: " + job.getJobName() + " on table " + tableName); - - IIManager iiMgr = IIManager.getInstance(KylinConfig.getInstanceFromEnv()); - IIInstance ii = iiMgr.getII(iiName); - job.getConfiguration().set(BatchConstants.TABLE_NAME, tableName); - job.getConfiguration().set(BatchConstants.TABLE_COLUMNS, getColumns(ii)); - - setJobClasspath(job); - - - setupMapper(ii.getFirstSegment()); - setupReducer(output); - - Configuration conf = job.getConfiguration(); - conf.set(BatchConstants.CFG_II_NAME, ii.getName()); - attachKylinPropsAndMetadata(ii, conf); - return waitForCompletion(job); - - } catch (Exception e) { - printUsage(options); - throw e; - } - - } - - private String getColumns(IIInstance ii) { - IIJoinedFlatTableDesc iiflat = new IIJoinedFlatTableDesc(ii.getDescriptor()); - StringBuilder buf = new StringBuilder(); - for (IntermediateColumnDesc col : iiflat.getColumnList()) { - if (buf.length() > 0) - buf.append(","); - buf.append(col.getColumnName()); - } - return buf.toString(); - } - - private void setupMapper(IISegment segment) throws IOException { - - IMRInput.IMRTableInputFormat flatTableInputFormat = MRUtil.getBatchCubingInputSide(segment).getFlatTableInputFormat(); - flatTableInputFormat.configureJob(job); - - job.setMapperClass(IIDistinctColumnsMapper.class); - job.setCombinerClass(IIDistinctColumnsCombiner.class); - job.setMapOutputKeyClass(ShortWritable.class); - job.setMapOutputValueClass(Text.class); - } - - private void setupReducer(Path output) throws IOException { - job.setReducerClass(IIDistinctColumnsReducer.class); - job.setOutputFormatClass(SequenceFileOutputFormat.class); - job.setOutputKeyClass(NullWritable.class); - job.setOutputValueClass(Text.class); - - FileOutputFormat.setOutputPath(job, output); - job.getConfiguration().set(BatchConstants.OUTPUT_PATH, output.toString()); - - job.setNumReduceTasks(1); - - deletePath(job.getConfiguration(), output); - } - - public static void main(String[] args) throws Exception { - IIDistinctColumnsJob job = new IIDistinctColumnsJob(); - int exitCode = ToolRunner.run(job, args); - System.exit(exitCode); - } -} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsMapper.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsMapper.java deleted file mode 100644 index c431ecd..0000000 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsMapper.java +++ /dev/null @@ -1,75 +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.invertedindex; - -import java.io.IOException; - -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.io.ShortWritable; -import org.apache.hadoop.io.Text; -import org.apache.kylin.common.KylinConfig; -import org.apache.kylin.common.util.Bytes; -import org.apache.kylin.engine.mr.IMRInput; -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.invertedindex.IIInstance; -import org.apache.kylin.invertedindex.IIManager; -import org.apache.kylin.invertedindex.IISegment; - -/** - * @author yangli9 - */ -public class IIDistinctColumnsMapper<KEYIN> extends KylinMapper<KEYIN, Object, ShortWritable, Text> { - - private ShortWritable outputKey = new ShortWritable(); - private Text outputValue = new Text(); - - protected IMRInput.IMRTableInputFormat flatTableInputFormat; - - @Override - protected void setup(Context context) throws IOException { - super.bindCurrentConfiguration(context.getConfiguration()); - Configuration conf = context.getConfiguration(); - KylinConfig config = AbstractHadoopJob.loadKylinPropsAndMetadata(); - - String iiName = conf.get(BatchConstants.CFG_II_NAME); - IIInstance iiInstance = IIManager.getInstance(config).getII(iiName); - IISegment seg = iiInstance.getFirstSegment(); - flatTableInputFormat = MRUtil.getBatchCubingInputSide(seg).getFlatTableInputFormat(); - } - - @Override - public void map(KEYIN key, Object record, Context context) throws IOException, InterruptedException { - - String[] row = flatTableInputFormat.parseMapperInput(record); - - for (short i = 0; i < row.length; i++) { - outputKey.set(i); - if (row[i] == null) - continue; - byte[] bytes = Bytes.toBytes(row[i].toString()); - outputValue.set(bytes, 0, bytes.length); - context.write(outputKey, outputValue); - } - - } - -} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsReducer.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsReducer.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsReducer.java deleted file mode 100644 index d50385f..0000000 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIDistinctColumnsReducer.java +++ /dev/null @@ -1,77 +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.invertedindex; - -import java.io.IOException; -import java.util.HashSet; - -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.FSDataOutputStream; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.io.NullWritable; -import org.apache.hadoop.io.ShortWritable; -import org.apache.hadoop.io.Text; -import org.apache.kylin.common.util.ByteArray; -import org.apache.kylin.common.util.Bytes; -import org.apache.kylin.engine.mr.KylinReducer; -import org.apache.kylin.engine.mr.common.BatchConstants; - -/** - * @author yangli9 - */ -public class IIDistinctColumnsReducer extends KylinReducer<ShortWritable, Text, NullWritable, Text> { - - private String[] columns; - - @Override - protected void setup(Context context) throws IOException { - super.bindCurrentConfiguration(context.getConfiguration()); - - Configuration conf = context.getConfiguration(); - this.columns = conf.get(BatchConstants.TABLE_COLUMNS).split(","); - } - - @Override - public void reduce(ShortWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException { - String columnName = columns[key.get()]; - - HashSet<ByteArray> set = new HashSet<ByteArray>(); - for (Text textValue : values) { - ByteArray value = new ByteArray(Bytes.copy(textValue.getBytes(), 0, textValue.getLength())); - set.add(value); - } - - Configuration conf = context.getConfiguration(); - FileSystem fs = FileSystem.get(conf); - String outputPath = conf.get(BatchConstants.OUTPUT_PATH); - FSDataOutputStream out = fs.create(new Path(outputPath, columnName)); - - try { - for (ByteArray value : set) { - out.write(value.array(), value.offset(), value.length()); - out.write('\n'); - } - } finally { - out.close(); - } - - } - -} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIJobBuilder.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIJobBuilder.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIJobBuilder.java deleted file mode 100644 index 18d3001..0000000 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/IIJobBuilder.java +++ /dev/null @@ -1,219 +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.invertedindex; -// -//import java.io.IOException; -//import java.text.SimpleDateFormat; -//import java.util.Date; -//import java.util.TimeZone; -// -//import org.apache.kylin.engine.mr.common.HadoopShellExecutable; -//import org.apache.kylin.engine.mr.common.MapReduceExecutable; -//import org.apache.kylin.invertedindex.IISegment; -//import org.apache.kylin.invertedindex.model.IIJoinedFlatTableDesc; -//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.metadata.model.DataModelDesc.RealizationCapacity; -// -//import com.google.common.base.Preconditions; -// -///** -// */ -//public final class IIJobBuilder { -// -// final JobEngineConfig engineConfig; -// -// public IIJobBuilder(JobEngineConfig engineConfig) { -// this.engineConfig = engineConfig; -// } -// -// public IIJob buildJob(IISegment seg, String submitter) { -// checkPreconditions(seg); -// -// IIJob result = initialJob(seg, "BUILD", submitter); -// final String jobId = result.getId(); -// final IIJoinedFlatTableDesc intermediateTableDesc = new IIJoinedFlatTableDesc(seg.getIIDesc()); -// final String intermediateTableIdentity = getIntermediateTableIdentity(intermediateTableDesc); -// final String factDistinctColumnsPath = getIIDistinctColumnsPath(seg, jobId); -// final String iiRootPath = getJobWorkingDir(jobId) + "/" + seg.getIIInstance().getName() + "/"; -// final String iiPath = iiRootPath + "*"; -// -// final AbstractExecutable intermediateHiveTableStep = createFlatHiveTableStep(intermediateTableDesc, jobId); -// result.addTask(intermediateHiveTableStep); -// -// result.addTask(createFactDistinctColumnsStep(seg, intermediateTableIdentity, jobId, factDistinctColumnsPath)); -// -// result.addTask(createBuildDictionaryStep(seg, factDistinctColumnsPath)); -// -// result.addTask(createInvertedIndexStep(seg, intermediateTableIdentity, iiRootPath)); -// -// // create htable step -// result.addTask(createCreateHTableStep(seg)); -// -// // generate hfiles step -// result.addTask(createConvertToHfileStep(seg, iiPath, jobId)); -// -// // bulk load step -// result.addTask(createBulkLoadStep(seg, jobId)); -// -// return result; -// } -// -// private IIJob initialJob(IISegment seg, String type, String submitter) { -// IIJob result = new IIJob(); -// SimpleDateFormat format = new SimpleDateFormat("z yyyy-MM-dd HH:mm:ss"); -// format.setTimeZone(TimeZone.getTimeZone(engineConfig.getTimeZone())); -// result.setIIName(seg.getIIInstance().getName()); -// result.setSegmentId(seg.getUuid()); -// result.setName(seg.getIIInstance().getName() + " - " + seg.getName() + " - " + type + " - " + format.format(new Date(System.currentTimeMillis()))); -// result.setSubmitter(submitter); -// return result; -// } -// -// private void checkPreconditions(IISegment seg) { -// Preconditions.checkNotNull(seg, "segment cannot be null"); -// Preconditions.checkNotNull(engineConfig, "jobEngineConfig cannot be null"); -// } -// -// private void appendMapReduceParameters(StringBuilder builder, JobEngineConfig engineConfig) { -// try { -// String jobConf = engineConfig.getHadoopJobConfFilePath(RealizationCapacity.MEDIUM); -// if (jobConf != null && jobConf.length() > 0) { -// builder.append(" -conf ").append(jobConf); -// } -// } catch (IOException e) { -// throw new RuntimeException(e); -// } -// } -// -// private String getIIDistinctColumnsPath(IISegment seg, String jobUuid) { -// return getJobWorkingDir(jobUuid) + "/" + seg.getIIInstance().getName() + "/ii_distinct_columns"; -// } -// -// private String getHFilePath(IISegment seg, String jobId) { -// return getJobWorkingDir(jobId) + "/" + seg.getIIInstance().getName() + "/hfile/"; -// } -// -// private MapReduceExecutable createFactDistinctColumnsStep(IISegment seg, String factTableName, String jobId, String output) { -// MapReduceExecutable result = new MapReduceExecutable(); -// result.setName(ExecutableConstants.STEP_NAME_FACT_DISTINCT_COLUMNS); -// result.setMapReduceJobClass(IIDistinctColumnsJob.class); -// StringBuilder cmd = new StringBuilder(); -// appendMapReduceParameters(cmd, engineConfig); -// appendExecCmdParameters(cmd, "tablename", factTableName); -// appendExecCmdParameters(cmd, "iiname", seg.getIIInstance().getName()); -// appendExecCmdParameters(cmd, "output", output); -// appendExecCmdParameters(cmd, "jobname", "Kylin_Fact_Distinct_Columns_" + seg.getIIInstance().getName() + "_Step"); -// -// result.setMapReduceParams(cmd.toString()); -// return result; -// } -// -// private HadoopShellExecutable createBuildDictionaryStep(IISegment seg, String factDistinctColumnsPath) { -// // base cuboid job -// HadoopShellExecutable buildDictionaryStep = new HadoopShellExecutable(); -// buildDictionaryStep.setName(ExecutableConstants.STEP_NAME_BUILD_DICTIONARY); -// StringBuilder cmd = new StringBuilder(); -// appendExecCmdParameters(cmd, "iiname", seg.getIIInstance().getName()); -// appendExecCmdParameters(cmd, "input", factDistinctColumnsPath); -// -// buildDictionaryStep.setJobParams(cmd.toString()); -// buildDictionaryStep.setJobClass(CreateInvertedIndexDictionaryJob.class); -// return buildDictionaryStep; -// } -// -// private MapReduceExecutable createInvertedIndexStep(IISegment seg, String intermediateHiveTable, String iiOutputTempPath) { -// // base cuboid job -// MapReduceExecutable buildIIStep = new MapReduceExecutable(); -// -// StringBuilder cmd = new StringBuilder(); -// appendMapReduceParameters(cmd, engineConfig); -// -// buildIIStep.setName(ExecutableConstants.STEP_NAME_BUILD_II); -// -// appendExecCmdParameters(cmd, "iiname", seg.getIIInstance().getName()); -// appendExecCmdParameters(cmd, "tablename", intermediateHiveTable); -// appendExecCmdParameters(cmd, "output", iiOutputTempPath); -// appendExecCmdParameters(cmd, "jobname", ExecutableConstants.STEP_NAME_BUILD_II); -// -// buildIIStep.setMapReduceParams(cmd.toString()); -// buildIIStep.setMapReduceJobClass(InvertedIndexJob.class); -// return buildIIStep; -// } -// -// private HadoopShellExecutable createCreateHTableStep(IISegment seg) { -// HadoopShellExecutable createHtableStep = new HadoopShellExecutable(); -// createHtableStep.setName(ExecutableConstants.STEP_NAME_CREATE_HBASE_TABLE); -// StringBuilder cmd = new StringBuilder(); -// appendExecCmdParameters(cmd, "iiname", seg.getIIInstance().getName()); -// appendExecCmdParameters(cmd, "htablename", seg.getStorageLocationIdentifier()); -// -// createHtableStep.setJobParams(cmd.toString()); -// createHtableStep.setJobClass(IICreateHTableJob.class); -// -// return createHtableStep; -// } -// -// private MapReduceExecutable createConvertToHfileStep(IISegment seg, String inputPath, String jobId) { -// MapReduceExecutable createHFilesStep = new MapReduceExecutable(); -// createHFilesStep.setName(ExecutableConstants.STEP_NAME_CONVERT_II_TO_HFILE); -// StringBuilder cmd = new StringBuilder(); -// -// appendMapReduceParameters(cmd, engineConfig); -// appendExecCmdParameters(cmd, "iiname", seg.getIIInstance().getName()); -// appendExecCmdParameters(cmd, "input", inputPath); -// appendExecCmdParameters(cmd, "output", getHFilePath(seg, jobId)); -// appendExecCmdParameters(cmd, "htablename", seg.getStorageLocationIdentifier()); -// appendExecCmdParameters(cmd, "jobname", "Kylin_HFile_Generator_" + seg.getIIInstance().getName() + "_Step"); -// -// createHFilesStep.setMapReduceParams(cmd.toString()); -// createHFilesStep.setMapReduceJobClass(IICreateHFileJob.class); -// -// return createHFilesStep; -// } -// -// private HadoopShellExecutable createBulkLoadStep(IISegment seg, String jobId) { -// HadoopShellExecutable bulkLoadStep = new HadoopShellExecutable(); -// bulkLoadStep.setName(ExecutableConstants.STEP_NAME_BULK_LOAD_HFILE); -// -// StringBuilder cmd = new StringBuilder(); -// appendExecCmdParameters(cmd, "input", getHFilePath(seg, jobId)); -// appendExecCmdParameters(cmd, "htablename", seg.getStorageLocationIdentifier()); -// appendExecCmdParameters(cmd, "iiname", seg.getIIInstance().getName()); -// -// bulkLoadStep.setJobParams(cmd.toString()); -// bulkLoadStep.setJobClass(IIBulkLoadJob.class); -// -// return bulkLoadStep; -// -// } -// -// private StringBuilder appendExecCmdParameters(StringBuilder buf, String paraName, String paraValue) { -// return buf.append(" -").append(paraName).append(" ").append(paraValue); -// } -// -// private String getJobWorkingDir(String uuid) { -// return engineConfig.getHdfsWorkingDirectory() + "kylin-" + uuid; -// } -// -// private String getIntermediateTableIdentity(IIJoinedFlatTableDesc intermediateTableDesc) { -// return engineConfig.getConfig().getHiveDatabaseForIntermediateTable() + "." + intermediateTableDesc.getTableName(); -// } -//} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexJob.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexJob.java index bcae524..5191aca 100644 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexJob.java +++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexJob.java @@ -26,6 +26,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.io.Writable; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; @@ -44,7 +45,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * @author yangli9 */ public class InvertedIndexJob extends AbstractHadoopJob { protected static final Logger logger = LoggerFactory.getLogger(InvertedIndexJob.class); @@ -56,13 +56,11 @@ public class InvertedIndexJob extends AbstractHadoopJob { try { options.addOption(OPTION_JOB_NAME); options.addOption(OPTION_II_NAME); - options.addOption(OPTION_TABLE_NAME); options.addOption(OPTION_OUTPUT_PATH); parseOptions(options, args); job = Job.getInstance(getConf(), getOptionValue(OPTION_JOB_NAME)); String iiname = getOptionValue(OPTION_II_NAME); - String intermediateTable = getOptionValue(OPTION_TABLE_NAME); Path output = new Path(getOptionValue(OPTION_OUTPUT_PATH)); // ---------------------------------------------------------------------------- @@ -111,17 +109,11 @@ public class InvertedIndexJob extends AbstractHadoopJob { private void setupMapper(IISegment segment) throws IOException { -// String[] dbTableNames = HadoopUtil.parseHiveTableName(intermediateTable); -// HCatInputFormat.setInput(job, dbTableNames[0], dbTableNames[1]); -// -// job.setInputFormatClass(HCatInputFormat.class); IMRInput.IMRTableInputFormat flatTableInputFormat = MRUtil.getBatchCubingInputSide(segment).getFlatTableInputFormat(); flatTableInputFormat.configureJob(job); - job.setMapperClass(InvertedIndexMapper.class); job.setMapOutputKeyClass(LongWritable.class); - job.setMapOutputValueClass(ImmutableBytesWritable.class); job.setPartitionerClass(InvertedIndexPartitioner.class); } http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexMapper.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexMapper.java index 88249ed..a1251a3 100644 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexMapper.java +++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexMapper.java @@ -18,13 +18,11 @@ package org.apache.kylin.engine.mr.invertedindex; -import java.io.IOException; -import java.util.List; - import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.io.Writable; import org.apache.kylin.common.KylinConfig; +import org.apache.kylin.common.util.DateFormat; import org.apache.kylin.engine.mr.IMRInput; import org.apache.kylin.engine.mr.KylinMapper; import org.apache.kylin.engine.mr.MRUtil; @@ -33,20 +31,19 @@ import org.apache.kylin.engine.mr.common.BatchConstants; import org.apache.kylin.invertedindex.IIInstance; import org.apache.kylin.invertedindex.IIManager; import org.apache.kylin.invertedindex.IISegment; -import org.apache.kylin.invertedindex.index.TableRecord; import org.apache.kylin.invertedindex.index.TableRecordInfo; import org.apache.kylin.metadata.model.SegmentStatusEnum; +import java.io.IOException; + /** * @author yangli9 */ -public class InvertedIndexMapper<KEYIN> extends KylinMapper<KEYIN, Object, LongWritable, ImmutableBytesWritable> { +public class InvertedIndexMapper<KEYIN> extends KylinMapper<KEYIN, Object, LongWritable, Writable> { private TableRecordInfo info; - private TableRecord rec; private LongWritable outputKey; - private ImmutableBytesWritable outputValue; private IMRInput.IMRTableInputFormat flatTableInputFormat; @Override @@ -60,10 +57,8 @@ public class InvertedIndexMapper<KEYIN> extends KylinMapper<KEYIN, Object, LongW IIInstance ii = mgr.getII(conf.get(BatchConstants.CFG_II_NAME)); IISegment seg = ii.getSegment(conf.get(BatchConstants.CFG_II_SEGMENT_NAME), SegmentStatusEnum.NEW); this.info = new TableRecordInfo(seg); - this.rec = this.info.createTableRecord(); outputKey = new LongWritable(); - outputValue = new ImmutableBytesWritable(rec.getBytes()); flatTableInputFormat = MRUtil.getBatchCubingInputSide(ii.getFirstSegment()).getFlatTableInputFormat(); } @@ -71,17 +66,12 @@ public class InvertedIndexMapper<KEYIN> extends KylinMapper<KEYIN, Object, LongW @Override public void map(KEYIN key, Object record, Context context) throws IOException, InterruptedException { - String[] row = flatTableInputFormat.parseMapperInput(record); - rec.reset(); - for (int i = 0; i < row.length; i++) { - Object fieldValue = row[i]; - if (fieldValue != null) - rec.setValueString(i, fieldValue.toString()); - } - - outputKey.set(rec.getTimestamp()); - // outputValue's backing bytes array is the same as rec + Writable writableRecord = (Writable) record; + String[] row = flatTableInputFormat.parseMapperInput(writableRecord); + String timestampString = row[info.getTimestampColumn()]; - context.write(outputKey, outputValue); + outputKey.set(DateFormat.stringToMillis(timestampString)); + // + context.write(outputKey, writableRecord); } } http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexReducer.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexReducer.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexReducer.java index 7644456..56c0b9e 100644 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexReducer.java +++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/InvertedIndexReducer.java @@ -18,35 +18,45 @@ package org.apache.kylin.engine.mr.invertedindex; -import java.io.IOException; - +import com.google.common.collect.Lists; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.io.LongWritable; import org.apache.kylin.common.KylinConfig; +import org.apache.kylin.common.util.Pair; +import org.apache.kylin.common.util.StreamingBatch; +import org.apache.kylin.common.util.StreamingMessage; +import org.apache.kylin.engine.mr.IMRInput; import org.apache.kylin.engine.mr.KylinReducer; +import org.apache.kylin.engine.mr.MRUtil; import org.apache.kylin.engine.mr.common.AbstractHadoopJob; import org.apache.kylin.engine.mr.common.BatchConstants; import org.apache.kylin.invertedindex.IIInstance; import org.apache.kylin.invertedindex.IIManager; import org.apache.kylin.invertedindex.IISegment; -import org.apache.kylin.invertedindex.index.IncrementalSliceMaker; import org.apache.kylin.invertedindex.index.Slice; -import org.apache.kylin.invertedindex.index.TableRecord; +import org.apache.kylin.invertedindex.index.SliceBuilder; import org.apache.kylin.invertedindex.index.TableRecordInfo; import org.apache.kylin.invertedindex.model.IIKeyValueCodec; import org.apache.kylin.invertedindex.model.IIRow; -import org.apache.kylin.metadata.model.SegmentStatusEnum; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; /** - * @author yangli9 */ -public class InvertedIndexReducer extends KylinReducer<LongWritable, ImmutableBytesWritable, ImmutableBytesWritable, ImmutableBytesWritable> { +public class InvertedIndexReducer extends KylinReducer<LongWritable, Object, ImmutableBytesWritable, ImmutableBytesWritable> { private TableRecordInfo info; - private TableRecord rec; - private IncrementalSliceMaker builder; private IIKeyValueCodec kv; + private IMRInput.IMRTableInputFormat flatTableInputFormat; + private SliceBuilder sliceBuilder; + private ArrayList<StreamingMessage> messages; + private int sliceSize; + private ImmutableBytesWritable immutableBytesWritable; + private ByteBuffer valueBuf; @Override protected void setup(Context context) throws IOException { @@ -56,44 +66,61 @@ public class InvertedIndexReducer extends KylinReducer<LongWritable, ImmutableBy KylinConfig config = AbstractHadoopJob.loadKylinPropsAndMetadata(); IIManager mgr = IIManager.getInstance(config); IIInstance ii = mgr.getII(conf.get(BatchConstants.CFG_II_NAME)); - IISegment seg = ii.getSegment(conf.get(BatchConstants.CFG_II_SEGMENT_NAME), SegmentStatusEnum.NEW); + IISegment seg = ii.getFirstSegment(); info = new TableRecordInfo(seg); - rec = info.createTableRecord(); - builder = null; kv = new IIKeyValueCodec(info.getDigest()); + flatTableInputFormat = MRUtil.getBatchCubingInputSide(ii.getFirstSegment()).getFlatTableInputFormat(); + sliceSize = ii.getDescriptor().getSliceSize(); + short shard = (short) context.getTaskAttemptID().getTaskID().getId(); + System.out.println("Generating to shard - " + shard); + sliceBuilder = new SliceBuilder(seg.getIIDesc(), shard, true); + messages = Lists.newArrayListWithCapacity(sliceSize); + immutableBytesWritable = new ImmutableBytesWritable(); + valueBuf = ByteBuffer.allocate(1024 * 1024); // 1MB } @Override - public void reduce(LongWritable key, Iterable<ImmutableBytesWritable> values, Context context) // + public void reduce(LongWritable key, Iterable<Object> values, Context context) // throws IOException, InterruptedException { - for (ImmutableBytesWritable v : values) { - rec.setBytes(v.get(), v.getOffset(), v.getLength()); - - if (builder == null) { - builder = new IncrementalSliceMaker(info, rec.getShard()); - } - - //TODO: to delete this log - System.out.println(rec.getShard() + " - " + rec); - - Slice slice = builder.append(rec); - if (slice != null) { - output(slice, context); + for (Object v : values) { + String[] row = flatTableInputFormat.parseMapperInput(v); + messages.add((parse(row))); + if (messages.size() >= sliceSize) { + buildAndOutput(new StreamingBatch(messages, Pair.newPair(System.currentTimeMillis(), System.currentTimeMillis())), context); + messages = Lists.newArrayList(); } } } + private StreamingMessage parse(String[] row) { + return new StreamingMessage(Lists.newArrayList(row), System.currentTimeMillis(), System.currentTimeMillis(), Collections.<String, Object> emptyMap()); + } + @Override protected void cleanup(Context context) throws IOException, InterruptedException { - Slice slice = builder.close(); - if (slice != null) { - output(slice, context); + if (!messages.isEmpty()) { + buildAndOutput(new StreamingBatch(messages, Pair.newPair(System.currentTimeMillis(), System.currentTimeMillis())), context); + messages.clear(); } + } - private void output(Slice slice, Context context) throws IOException, InterruptedException { + private void buildAndOutput(StreamingBatch streamingBatch, Context context) throws IOException, InterruptedException { + final Slice slice = sliceBuilder.buildSlice(streamingBatch); + ImmutableBytesWritable value, dictionary; for (IIRow pair : kv.encodeKeyValue(slice)) { - context.write(pair.getKey(), pair.getValue()); + value = pair.getValue(); + dictionary = pair.getDictionary(); + int newLength = 4 + value.getLength() + dictionary.getLength(); + if (newLength > valueBuf.limit()) { + valueBuf = ByteBuffer.allocate(newLength); + } + valueBuf.clear(); + valueBuf.putInt(value.getLength()); + valueBuf.put(value.get(), value.getOffset(), value.getLength()); + valueBuf.put(dictionary.get(), dictionary.getOffset(), dictionary.getLength()); + immutableBytesWritable.set(valueBuf.array(), 0, newLength); + context.write(pair.getKey(), immutableBytesWritable); } } http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/UpdateIIInfoAfterBuildStep.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/UpdateIIInfoAfterBuildStep.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/UpdateIIInfoAfterBuildStep.java new file mode 100644 index 0000000..ef9b1c3 --- /dev/null +++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/UpdateIIInfoAfterBuildStep.java @@ -0,0 +1,84 @@ +/* + * 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.invertedindex; + +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.invertedindex.IIInstance; +import org.apache.kylin.invertedindex.IIManager; +import org.apache.kylin.invertedindex.IISegment; +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.SegmentStatusEnum; + +import java.io.IOException; + +/** + */ +public class UpdateIIInfoAfterBuildStep extends AbstractExecutable { + + private static final String II_NAME = "iiName"; + private static final String JOB_ID = "jobId"; + + public UpdateIIInfoAfterBuildStep() { + super(); + } + + public void setInvertedIndexName(String cubeName) { + this.setParam(II_NAME, cubeName); + } + + private String getInvertedIndexName() { + return getParam(II_NAME); + } + + public void setJobId(String id) { + setParam(JOB_ID, id); + } + + private String getJobId() { + return getParam(JOB_ID); + } + + @Override + protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException { + + IIManager mgr = IIManager.getInstance(KylinConfig.getInstanceFromEnv()); + IIInstance ii = mgr.getII(getInvertedIndexName()); + IISegment segment = ii.getFirstSegment(); + segment.setStatus(SegmentStatusEnum.READY); + + segment.setLastBuildJobID(getJobId()); + segment.setLastBuildTime(System.currentTimeMillis()); + + try { + mgr.updateII(ii); + return new ExecuteResult(ExecuteResult.State.SUCCEED, "succeed"); + } catch (IOException e) { + logger.error("fail to update inverted index after build", e); + return new ExecuteResult(ExecuteResult.State.ERROR, e.getLocalizedMessage()); + } + } + +} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/UpdateInvertedIndexInfoAfterBuildStep.java ---------------------------------------------------------------------- diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/UpdateInvertedIndexInfoAfterBuildStep.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/UpdateInvertedIndexInfoAfterBuildStep.java deleted file mode 100644 index 277dea5..0000000 --- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/invertedindex/UpdateInvertedIndexInfoAfterBuildStep.java +++ /dev/null @@ -1,93 +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.invertedindex; - -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.invertedindex.IIInstance; -import org.apache.kylin.invertedindex.IIManager; -import org.apache.kylin.invertedindex.IISegment; -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.SegmentStatusEnum; - -import java.io.IOException; - -/** - */ -public class UpdateInvertedIndexInfoAfterBuildStep extends AbstractExecutable { - - private static final String SEGMENT_ID = "segmentId"; - private static final String II_NAME = "iiName"; - private static final String JOB_ID = "jobId"; - - public UpdateInvertedIndexInfoAfterBuildStep() { - super(); - } - - public void setInvertedIndexName(String cubeName) { - this.setParam(II_NAME, cubeName); - } - - private String getInvertedIndexName() { - return getParam(II_NAME); - } - - public void setSegmentId(String segmentId) { - this.setParam(SEGMENT_ID, segmentId); - } - - private String getSegmentId() { - return getParam(SEGMENT_ID); - } - - public void setJobId(String id) { - setParam(JOB_ID, id); - } - - private String getJobId() { - return getParam(JOB_ID); - } - - @Override - protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException { - - IIManager mgr = IIManager.getInstance(KylinConfig.getInstanceFromEnv()); - IIInstance ii = mgr.getII(getInvertedIndexName()); - IISegment segment = ii.getFirstSegment(); - segment.setStatus(SegmentStatusEnum.READY); - - segment.setLastBuildJobID(getJobId()); - segment.setLastBuildTime(System.currentTimeMillis()); - - try { - mgr.updateII(ii); - return new ExecuteResult(ExecuteResult.State.SUCCEED, "succeed"); - } catch (IOException e) { - logger.error("fail to update inverted index after build", e); - return new ExecuteResult(ExecuteResult.State.ERROR, e.getLocalizedMessage()); - } - } - -} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/IStreamingInput.java ---------------------------------------------------------------------- diff --git a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/IStreamingInput.java b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/IStreamingInput.java index d8cb24b..1cf3d98 100644 --- a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/IStreamingInput.java +++ b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/IStreamingInput.java @@ -33,6 +33,8 @@ */ package org.apache.kylin.engine.streaming; +import org.apache.kylin.common.util.StreamingBatch; + /** */ public interface IStreamingInput { http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/OneOffStreamingBuilder.java ---------------------------------------------------------------------- diff --git a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/OneOffStreamingBuilder.java b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/OneOffStreamingBuilder.java index 475e43a..4a3b8b8 100644 --- a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/OneOffStreamingBuilder.java +++ b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/OneOffStreamingBuilder.java @@ -36,6 +36,7 @@ package org.apache.kylin.engine.streaming; import java.util.Map; import org.apache.kylin.common.hll.HyperLogLogPlusCounter; +import org.apache.kylin.common.util.StreamingBatch; import org.apache.kylin.dict.Dictionary; import org.apache.kylin.engine.streaming.util.StreamingUtils; import org.apache.kylin.metadata.model.IBuildable; http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingBatch.java ---------------------------------------------------------------------- diff --git a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingBatch.java b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingBatch.java deleted file mode 100644 index c7c7d29..0000000 --- a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingBatch.java +++ /dev/null @@ -1,61 +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.streaming; - -import java.util.List; - -import org.apache.kylin.common.util.Pair; - -/** - */ -public final class StreamingBatch { - - private final List<StreamingMessage> messages; - - private final Pair<Long, Long> timeRange; - - public StreamingBatch(List<StreamingMessage> messages, Pair<Long, Long> timeRange) { - this.messages = messages; - this.timeRange = timeRange; - } - - public List<StreamingMessage> getMessages() { - return messages; - } - - public Pair<Long, Long> getTimeRange() { - return timeRange; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingBatchBuilder.java ---------------------------------------------------------------------- diff --git a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingBatchBuilder.java b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingBatchBuilder.java index d55ccdb..93cda2d 100644 --- a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingBatchBuilder.java +++ b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingBatchBuilder.java @@ -36,6 +36,7 @@ package org.apache.kylin.engine.streaming; import java.util.Map; import org.apache.kylin.common.hll.HyperLogLogPlusCounter; +import org.apache.kylin.common.util.StreamingBatch; import org.apache.kylin.cube.inmemcubing.ICuboidWriter; import org.apache.kylin.dict.Dictionary; import org.apache.kylin.metadata.model.IBuildable; http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingMessage.java ---------------------------------------------------------------------- diff --git a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingMessage.java b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingMessage.java deleted file mode 100644 index 7885902..0000000 --- a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/StreamingMessage.java +++ /dev/null @@ -1,43 +0,0 @@ -package org.apache.kylin.engine.streaming; - -import java.util.Collections; -import java.util.List; -import java.util.Map; - -/** - */ -public class StreamingMessage { - - private final List<String> data; - - private long offset; - - private long timestamp; - - private Map<String, Object> params; - - public static final StreamingMessage EOF = new StreamingMessage(Collections.<String> emptyList(), 0L, 0L, Collections.<String, Object> emptyMap()); - - public StreamingMessage(List<String> data, long offset, long timestamp, Map<String, Object> params) { - this.data = data; - this.offset = offset; - this.timestamp = timestamp; - this.params = params; - } - - public final List<String> getData() { - return data; - } - - public final long getOffset() { - return offset; - } - - public final long getTimestamp() { - return timestamp; - } - - public Map<String, Object> getParams() { - return params; - } -} http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/cube/StreamingCubeBuilder.java ---------------------------------------------------------------------- diff --git a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/cube/StreamingCubeBuilder.java b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/cube/StreamingCubeBuilder.java index 64a3061..ae72218 100644 --- a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/cube/StreamingCubeBuilder.java +++ b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/cube/StreamingCubeBuilder.java @@ -54,9 +54,9 @@ import org.apache.kylin.cube.inmemcubing.ICuboidWriter; import org.apache.kylin.cube.inmemcubing.InMemCubeBuilder; import org.apache.kylin.cube.util.CubingUtils; import org.apache.kylin.dict.Dictionary; -import org.apache.kylin.engine.streaming.StreamingBatch; +import org.apache.kylin.common.util.StreamingBatch; import org.apache.kylin.engine.streaming.StreamingBatchBuilder; -import org.apache.kylin.engine.streaming.StreamingMessage; +import org.apache.kylin.common.util.StreamingMessage; import org.apache.kylin.metadata.model.IBuildable; import org.apache.kylin.metadata.model.SegmentStatusEnum; import org.apache.kylin.metadata.model.TblColRef; http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/82bfa924/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/invertedindex/SliceBuilder.java ---------------------------------------------------------------------- diff --git a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/invertedindex/SliceBuilder.java b/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/invertedindex/SliceBuilder.java deleted file mode 100644 index fa5a0b2..0000000 --- a/engine-streaming/src/main/java/org/apache/kylin/engine/streaming/invertedindex/SliceBuilder.java +++ /dev/null @@ -1,81 +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.streaming.invertedindex; - -import com.google.common.base.Function; -import com.google.common.collect.Lists; -import org.apache.kylin.dict.Dictionary; -import org.apache.kylin.engine.streaming.StreamingBatch; -import org.apache.kylin.engine.streaming.StreamingMessage; -import org.apache.kylin.invertedindex.index.BatchSliceMaker; -import org.apache.kylin.invertedindex.index.Slice; -import org.apache.kylin.invertedindex.index.TableRecord; -import org.apache.kylin.invertedindex.index.TableRecordInfo; -import org.apache.kylin.invertedindex.model.IIDesc; -import org.apache.kylin.invertedindex.util.IIDictionaryBuilder; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import javax.annotation.Nullable; -import java.util.List; - -/** - */ -public final class SliceBuilder { - - private static Logger logger = LoggerFactory.getLogger(SliceBuilder.class); - - private final BatchSliceMaker sliceMaker; - private final IIDesc iiDesc; - private final boolean useLocalDict; - - public SliceBuilder(IIDesc desc, short shard, boolean useLocalDict) { - this.iiDesc = desc; - this.sliceMaker = new BatchSliceMaker(desc, shard); - this.useLocalDict = useLocalDict; - } - - public Slice buildSlice(StreamingBatch microStreamBatch) { - final List<List<String>> messages = Lists.transform(microStreamBatch.getMessages(), new Function<StreamingMessage, List<String>>() { - @Nullable - @Override - public List<String> apply(@Nullable StreamingMessage input) { - return input.getData(); - } - }); - final Dictionary<?>[] dictionaries = useLocalDict ? IIDictionaryBuilder.buildDictionary(messages, iiDesc) : new Dictionary[iiDesc.listAllColumns().size()]; - TableRecordInfo tableRecordInfo = new TableRecordInfo(iiDesc, dictionaries); - return build(messages, tableRecordInfo, dictionaries); - } - - private Slice build(List<List<String>> table, final TableRecordInfo tableRecordInfo, Dictionary<?>[] localDictionary) { - final Slice slice = sliceMaker.makeSlice(tableRecordInfo.getDigest(), Lists.transform(table, new Function<List<String>, TableRecord>() { - @Nullable - @Override - public TableRecord apply(@Nullable List<String> input) { - TableRecord result = tableRecordInfo.createTableRecord(); - for (int i = 0; i < input.size(); i++) { - result.setValueString(i, input.get(i)); - } - return result; - } - })); - slice.setLocalDictionaries(localDictionary); - return slice; - } -}
