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 2016/11/27 10:28:12 UTC
[15/50] [abbrv] kylin git commit: KYLIN-2202 fix the conflict between
KYLIN-1851 and KYLIN-2202. Remove some useless MR files.
KYLIN-2202 fix the conflict between KYLIN-1851 and KYLIN-2202. Remove some useless MR files.
Signed-off-by: Li Yang <li...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/2b7fe610
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/2b7fe610
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/2b7fe610
Branch: refs/heads/KYLIN-1875
Commit: 2b7fe6106d5363d9993dfd7b7f27c49319ed7d44
Parents: 79cb59c
Author: xiefan46 <95...@qq.com>
Authored: Thu Nov 17 14:21:38 2016 +0800
Committer: Li Yang <li...@apache.org>
Committed: Tue Nov 22 18:39:35 2016 +0800
----------------------------------------------------------------------
.../kylin/cube/AggregationGroupRuleTest.java | 2 +-
.../validation/rule/DictionaryRuleTest.java | 1 +
.../apache/kylin/dict/DictionaryGenerator.java | 4 +-
.../apache/kylin/dict/DictionaryManager.java | 3 +
.../dict/NumberDictionaryForestBuilder.java | 5 +
.../dict/TableColumnValueSortedEnumerator.java | 154 ++
.../kylin/dict/TrieDictionaryForestBuilder.java | 2 +-
.../apache/kylin/dict/lookup/SnapshotTable.java | 2 -
.../kylin/dict/TrieDictionaryForestTest.java | 8 +-
.../org/apache/kylin/source/ReadableTable.java | 17 +-
.../apache/kylin/engine/mr/DFSFileTable.java | 36 +
.../engine/mr/DFSSingleFileTableReader.java | 218 ++
.../kylin/engine/mr/JobBuilderSupport.java | 4 +-
.../apache/kylin/engine/mr/SortedColumn.java | 100 +
.../kylin/engine/mr/SortedColumnReader.java | 136 ++
.../engine/mr/steps/CreateDictionaryJob.java | 5 +-
.../mr/steps/FactDistinctColumnPartitioner.java | 5 +-
.../mr/steps/FactDistinctColumnsCombiner.java | 4 +-
.../engine/mr/steps/FactDistinctColumnsJob.java | 2 +-
.../mr/steps/FactDistinctColumnsMapperBase.java | 3 +-
.../mr/steps/FactDistinctColumnsReducer.java | 6 +-
.../mr/steps/FactDistinctHiveColumnsMapper.java | 22 +-
.../engine/mr/steps/SelfDefineSortableKey.java | 128 ++
.../apache/kylin/engine/mr/steps/TypeFlag.java | 28 +
.../mr/steps/UpdateCubeInfoAfterBuildStep.java | 5 +-
.../fdc2/FactDistinctColumnPartitioner2.java | 47 -
.../fdc2/FactDistinctColumnsCombiner2.java | 45 -
.../mr/steps/fdc2/FactDistinctColumnsJob2.java | 148 --
.../fdc2/FactDistinctColumnsMapperBase2.java | 103 -
.../steps/fdc2/FactDistinctColumnsReducer2.java | 254 ---
.../fdc2/FactDistinctHiveColumnsMapper2.java | 232 --
.../mr/steps/fdc2/SelfDefineSortableKey.java | 131 --
.../kylin/engine/mr/steps/fdc2/TypeFlag.java | 28 -
.../kylin/engine/mr/SortedColumnReaderTest.java | 312 +++
.../mr/steps/NumberDictionaryForestTest.java | 17 +-
.../mr/steps/SelfDefineSortableKeyTest.java | 2 -
.../src/test/resources/multi_file_double/data_0 | 2006 +++++++++++++++++
.../src/test/resources/multi_file_double/data_1 | 2093 ++++++++++++++++++
.../src/test/resources/multi_file_double/data_2 | 2006 +++++++++++++++++
.../src/test/resources/multi_file_double/data_3 | 1947 ++++++++++++++++
.../src/test/resources/multi_file_double/data_4 | 1948 ++++++++++++++++
.../resources/multi_file_empty_file/empty.txt | 0
.../multi_file_empty_file/has_value.txt | 2 +
.../src/test/resources/multi_file_int/data_0 | 1987 +++++++++++++++++
.../src/test/resources/multi_file_int/data_1 | 1989 +++++++++++++++++
.../src/test/resources/multi_file_int/data_2 | 1998 +++++++++++++++++
.../src/test/resources/multi_file_int/data_3 | 1996 +++++++++++++++++
.../src/test/resources/multi_file_int/data_4 | 2030 +++++++++++++++++
.../src/test/resources/multi_file_str/data_0 | 2029 +++++++++++++++++
.../src/test/resources/multi_file_str/data_1 | 1990 +++++++++++++++++
.../src/test/resources/multi_file_str/data_2 | 1992 +++++++++++++++++
.../src/test/resources/multi_file_str/data_3 | 1948 ++++++++++++++++
.../src/test/resources/multi_file_str/data_4 | 2041 +++++++++++++++++
.../org/apache/kylin/source/hive/HiveTable.java | 3 +
54 files changed, 31200 insertions(+), 1024 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/core-cube/src/test/java/org/apache/kylin/cube/AggregationGroupRuleTest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/test/java/org/apache/kylin/cube/AggregationGroupRuleTest.java b/core-cube/src/test/java/org/apache/kylin/cube/AggregationGroupRuleTest.java
index 1c43093..1950b02 100644
--- a/core-cube/src/test/java/org/apache/kylin/cube/AggregationGroupRuleTest.java
+++ b/core-cube/src/test/java/org/apache/kylin/cube/AggregationGroupRuleTest.java
@@ -38,7 +38,7 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-public class AggregationGroupRuleTest extends LocalFileMetadataTestCase{
+public class AggregationGroupRuleTest extends LocalFileMetadataTestCase {
@Before
public void setUp() throws Exception {
this.createTestMetadata();
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/core-cube/src/test/java/org/apache/kylin/cube/model/validation/rule/DictionaryRuleTest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/test/java/org/apache/kylin/cube/model/validation/rule/DictionaryRuleTest.java b/core-cube/src/test/java/org/apache/kylin/cube/model/validation/rule/DictionaryRuleTest.java
index 9b37507..8bd4c88 100644
--- a/core-cube/src/test/java/org/apache/kylin/cube/model/validation/rule/DictionaryRuleTest.java
+++ b/core-cube/src/test/java/org/apache/kylin/cube/model/validation/rule/DictionaryRuleTest.java
@@ -34,6 +34,7 @@ import org.apache.kylin.cube.model.validation.ValidateContext;
import org.apache.kylin.dict.GlobalDictionaryBuilder;
import org.junit.After;
import org.junit.Before;
+import org.junit.Ignore;
import org.junit.Test;
/**
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryGenerator.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryGenerator.java b/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryGenerator.java
index 38af4bc..20a57ba 100644
--- a/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryGenerator.java
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryGenerator.java
@@ -153,7 +153,9 @@ public class DictionaryGenerator {
private static class NumberDictBuilder implements IDictionaryBuilder {
@Override
public Dictionary<String> build(DictionaryInfo dictInfo, IDictionaryValueEnumerator valueEnumerator, int baseId, int nSamples, ArrayList<String> returnSamples) throws IOException {
- NumberDictionaryForestBuilder builder = new NumberDictionaryForestBuilder(baseId);
+
+ int maxTrieSizeInMB = TrieDictionaryForestBuilder.getMaxTrieSizeInMB();
+ NumberDictionaryForestBuilder builder = new NumberDictionaryForestBuilder(baseId, maxTrieSizeInMB);
String value;
while (valueEnumerator.moveNext()) {
value = valueEnumerator.current();
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java b/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
index db165ba..0608e6f 100644
--- a/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
@@ -442,6 +442,9 @@ public class DictionaryManager {
logger.info("DictionaryManager(" + System.identityHashCode(this) + ") loading DictionaryInfo(loadDictObj:" + loadDictObj + ") at " + resourcePath);
DictionaryInfo info = store.getResource(resourcePath, DictionaryInfo.class, loadDictObj ? DictionaryInfoSerializer.FULL_SERIALIZER : DictionaryInfoSerializer.INFO_SERIALIZER);
+ //info.dictionaryObject.dump(System.out);
+ // if (loadDictObj)
+ // logger.debug("Loaded dictionary at " + resourcePath);
return info;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/core-dictionary/src/main/java/org/apache/kylin/dict/NumberDictionaryForestBuilder.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/NumberDictionaryForestBuilder.java b/core-dictionary/src/main/java/org/apache/kylin/dict/NumberDictionaryForestBuilder.java
index 4bd6c0f..a68b18e 100644
--- a/core-dictionary/src/main/java/org/apache/kylin/dict/NumberDictionaryForestBuilder.java
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/NumberDictionaryForestBuilder.java
@@ -20,12 +20,16 @@ package org.apache.kylin.dict;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.dict.NumberDictionary.NumberBytesCodec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Created by xiefan on 16-11-2.
*/
public class NumberDictionaryForestBuilder extends TrieDictionaryForestBuilder<String> {
+ private static final Logger logger = LoggerFactory.getLogger(NumberDictionaryForestBuilder.class);
+
public static class Number2BytesConverter implements BytesConverter<String> {
static final int MAX_DIGITS_BEFORE_DECIMAL_POINT = NumberDictionary.MAX_DIGITS_BEFORE_DECIMAL_POINT;
@@ -59,6 +63,7 @@ public class NumberDictionaryForestBuilder extends TrieDictionaryForestBuilder<S
codec.buf = backup;
return Bytes.toString(backup, 0, len);
}
+
}
public NumberDictionaryForestBuilder() {
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/core-dictionary/src/main/java/org/apache/kylin/dict/TableColumnValueSortedEnumerator.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/TableColumnValueSortedEnumerator.java b/core-dictionary/src/main/java/org/apache/kylin/dict/TableColumnValueSortedEnumerator.java
new file mode 100644
index 0000000..c57cc5e
--- /dev/null
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/TableColumnValueSortedEnumerator.java
@@ -0,0 +1,154 @@
+/*
+ * 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.dict;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.PriorityQueue;
+
+import org.apache.kylin.source.ReadableTable;
+
+/**
+ * Created by xiefan46 on 11/14/16.
+ */
+public class TableColumnValueSortedEnumerator implements IDictionaryValueEnumerator {
+
+ private Collection<ReadableTable.TableReader> readers;
+
+ private int colIndex;
+
+ private String colValue;
+
+ private Comparator<String> comparator;
+
+ private PriorityQueue<ReaderBuffer> pq;
+
+ public TableColumnValueSortedEnumerator(Collection<ReadableTable.TableReader> readers, int colIndex, final Comparator<String> comparator) {
+ this.readers = readers;
+ this.colIndex = colIndex;
+ this.comparator = comparator;
+ pq = new PriorityQueue<ReaderBuffer>(11, new Comparator<ReaderBuffer>() {
+ @Override
+ public int compare(ReaderBuffer i, ReaderBuffer j) {
+ boolean isEmpty1 = i.empty();
+ boolean isEmpty2 = j.empty();
+ if (isEmpty1 && isEmpty2)
+ return 0;
+ if (isEmpty1 && !isEmpty2)
+ return 1;
+ if (!isEmpty1 && isEmpty2)
+ return -1;
+ return comparator.compare(i.peek(), j.peek());
+ }
+ });
+ for (ReadableTable.TableReader reader : readers) {
+ if (reader != null) {
+ try {
+ pq.add(new ReaderBuffer(reader));
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+
+ @Override
+ public boolean moveNext() throws IOException {
+ while (pq.size() > 0) {
+ ReaderBuffer buffer = pq.poll();
+ String minEntry = buffer.pop();
+ this.colValue = minEntry;
+ if (buffer.empty()) {
+ pq.remove(buffer);
+ } else {
+ pq.add(buffer); // add it back
+ }
+ if (this.colValue == null) { //avoid the case of empty file
+ return false;
+ }
+ return true;
+ }
+ return false;
+ }
+
+
+ @Override
+ public void close() throws IOException {
+ for (ReadableTable.TableReader reader : readers) {
+ if (reader != null)
+ reader.close();
+ }
+ }
+
+ @Override
+ public String current() {
+ return colValue;
+ }
+
+ final class ReaderBuffer {
+ public ReaderBuffer(ReadableTable.TableReader reader) throws IOException {
+ this.reader = reader;
+ reload();
+ }
+
+ public void close() throws IOException {
+ if (this.reader != null)
+ reader.close();
+ }
+
+ public boolean empty() {
+ return (this.cache == null);
+ }
+
+ public String peek() {
+ return this.cache;
+ }
+
+ public String pop() throws IOException {
+ String result = this.cache;
+ reload();
+ return result;
+ }
+
+ private void reload() throws IOException {
+ if (reader.next()) {
+ String[] split = reader.getRow();
+ if (split.length == 1) {
+ this.cache = split[0];
+ } else {
+ // normal case
+ if (split.length <= colIndex) {
+ throw new ArrayIndexOutOfBoundsException("Column no. " + colIndex + " not found, line split is " + Arrays.asList(split));
+ }
+ this.cache = split[colIndex];
+ }
+
+ } else {
+ this.cache = null;
+ }
+ }
+
+ private String cache;
+
+ private ReadableTable.TableReader reader;
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/core-dictionary/src/main/java/org/apache/kylin/dict/TrieDictionaryForestBuilder.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/TrieDictionaryForestBuilder.java b/core-dictionary/src/main/java/org/apache/kylin/dict/TrieDictionaryForestBuilder.java
index 238f4b8..6aeb2ec 100755
--- a/core-dictionary/src/main/java/org/apache/kylin/dict/TrieDictionaryForestBuilder.java
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/TrieDictionaryForestBuilder.java
@@ -78,9 +78,9 @@ public class TrieDictionaryForestBuilder<T> {
addValue(valueBytes);
}
+
private void addValue(byte[] valueBytes) {
ByteArray valueByteArray = new ByteArray(valueBytes);
-
if (previousValue != null && isOrdered) {
int comp = previousValue.compareTo(valueByteArray);
if (comp == 0) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/core-dictionary/src/main/java/org/apache/kylin/dict/lookup/SnapshotTable.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/lookup/SnapshotTable.java b/core-dictionary/src/main/java/org/apache/kylin/dict/lookup/SnapshotTable.java
index 34b326a..db1a170 100644
--- a/core-dictionary/src/main/java/org/apache/kylin/dict/lookup/SnapshotTable.java
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/lookup/SnapshotTable.java
@@ -51,7 +51,6 @@ import com.fasterxml.jackson.annotation.JsonProperty;
@JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
public class SnapshotTable extends RootPersistentEntity implements ReadableTable {
-
@JsonProperty("tableName")
private String tableName;
@JsonProperty("signature")
@@ -59,7 +58,6 @@ public class SnapshotTable extends RootPersistentEntity implements ReadableTable
@JsonProperty("useDictionary")
private boolean useDictionary;
-
private ArrayList<int[]> rowIndices;
private Dictionary<String> dict;
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/core-dictionary/src/test/java/org/apache/kylin/dict/TrieDictionaryForestTest.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/test/java/org/apache/kylin/dict/TrieDictionaryForestTest.java b/core-dictionary/src/test/java/org/apache/kylin/dict/TrieDictionaryForestTest.java
index acd6894..32a6ab5 100755
--- a/core-dictionary/src/test/java/org/apache/kylin/dict/TrieDictionaryForestTest.java
+++ b/core-dictionary/src/test/java/org/apache/kylin/dict/TrieDictionaryForestTest.java
@@ -457,7 +457,7 @@ public class TrieDictionaryForestTest {
}
@Test
- public void testUnsortedData(){
+ public void testUnsortedData() {
ArrayList<String> strs = new ArrayList<>();
Iterator<String> it = new RandomStrings(10000).iterator();
int totalSize = 0;
@@ -477,11 +477,11 @@ public class TrieDictionaryForestTest {
//test maintain one trie
TrieDictionaryForestBuilder<String> builder = new TrieDictionaryForestBuilder<String>(converter);
builder.setMaxTrieTreeSize(maxTreeSize);
- for(String str : strs){
+ for (String str : strs) {
builder.addValue(str);
}
TrieDictionaryForest<String> dict = builder.build();
- assertEquals(1,dict.getTrees().size());
+ assertEquals(1, dict.getTrees().size());
//test throws Exception
Collections.sort(strs);
strs.add("f");
@@ -493,7 +493,7 @@ public class TrieDictionaryForestTest {
builder.addValue(str);
dict = builder.build();
fail("Input data no sorted and builder have multi trees. Should throw IllegalStateException");
- }catch (IllegalStateException e){
+ } catch (IllegalStateException e) {
//correct
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/core-metadata/src/main/java/org/apache/kylin/source/ReadableTable.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/source/ReadableTable.java b/core-metadata/src/main/java/org/apache/kylin/source/ReadableTable.java
index 0c40f6a..f127821 100644
--- a/core-metadata/src/main/java/org/apache/kylin/source/ReadableTable.java
+++ b/core-metadata/src/main/java/org/apache/kylin/source/ReadableTable.java
@@ -29,18 +29,27 @@ import com.fasterxml.jackson.annotation.JsonProperty;
*/
public interface ReadableTable {
- /** Returns a reader to read the table. */
+ /**
+ * Returns a reader to read the table.
+ */
public TableReader getReader() throws IOException;
- /** Used to detect table modifications mainly. Return null in case table does not exist. */
+ /**
+ * Used to detect table modifications mainly. Return null in case table does not exist.
+ */
public TableSignature getSignature() throws IOException;
+
public interface TableReader extends Closeable {
- /** Move to the next row, return false if no more record. */
+ /**
+ * Move to the next row, return false if no more record.
+ */
public boolean next() throws IOException;
- /** Get the current row. */
+ /**
+ * Get the current row.
+ */
public String[] getRow();
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/DFSFileTable.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/DFSFileTable.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/DFSFileTable.java
index f2b34fc..2193eab 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/DFSFileTable.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/DFSFileTable.java
@@ -18,10 +18,12 @@
package org.apache.kylin.engine.mr;
+import java.io.EOFException;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
@@ -55,6 +57,7 @@ public class DFSFileTable implements ReadableTable {
}
@Override
+ @Deprecated
public TableReader getReader() throws IOException {
return new DFSFileTableReader(path, delim, nColumns);
}
@@ -69,6 +72,29 @@ public class DFSFileTable implements ReadableTable {
}
}
+ public Collection<TableReader> getReaders() {
+ ArrayList<TableReader> readers = new ArrayList<>();
+ try {
+ String filePath = HadoopUtil.fixWindowsPath(path);
+ FileSystem fs = HadoopUtil.getFileSystem(filePath);
+ ArrayList<FileStatus> allFiles = new ArrayList<>();
+ FileStatus status = fs.getFileStatus(new Path(filePath));
+ if (status.isFile()) {
+ allFiles.add(status);
+ } else {
+ FileStatus[] listStatus = fs.listStatus(new Path(filePath));
+ allFiles.addAll(Arrays.asList(listStatus));
+ }
+ for (FileStatus f : allFiles) {
+ DFSSingleFileTableReader reader = new DFSSingleFileTableReader(f.getPath().toString(), delim, nColumns);
+ readers.add(reader);
+ }
+ }catch (IOException e){
+ e.printStackTrace();
+ }
+ return readers;
+ }
+
@Override
public String toString() {
return path;
@@ -96,4 +122,14 @@ public class DFSFileTable implements ReadableTable {
return Pair.newPair(size, lastModified);
}
+
+ private boolean isExceptionSayingNotSeqFile(IOException e) {
+ if (e.getMessage() != null && e.getMessage().contains("not a SequenceFile"))
+ return true;
+
+ if (e instanceof EOFException) // in case the file is very very small
+ return true;
+
+ return false;
+ }
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/DFSSingleFileTableReader.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/DFSSingleFileTableReader.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/DFSSingleFileTableReader.java
new file mode 100644
index 0000000..0a45188
--- /dev/null
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/DFSSingleFileTableReader.java
@@ -0,0 +1,218 @@
+/*
+ * 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.BufferedReader;
+import java.io.Closeable;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStreamReader;
+
+import org.apache.commons.lang.StringEscapeUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.SequenceFile.Reader;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.kylin.common.util.Bytes;
+import org.apache.kylin.common.util.StringSplitter;
+import org.apache.kylin.source.ReadableTable.TableReader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tables are typically CSV or SEQ file.
+ *
+ * @author yangli9
+ */
+public class DFSSingleFileTableReader implements TableReader {
+
+ private static final Logger logger = LoggerFactory.getLogger(DFSSingleFileTableReader.class);
+ private static final char CSV_QUOTE = '"';
+ private static final String[] DETECT_DELIMS = new String[] { "\177", "|", "\t", "," };
+
+ private String filePath;
+ private String delim;
+ private RowReader reader;
+
+ private String curLine;
+ private String[] curColumns;
+ private int expectedColumnNumber = -1; // helps delimiter detection
+
+ public DFSSingleFileTableReader(String filePath, int expectedColumnNumber) throws IOException {
+ this(filePath, DFSFileTable.DELIM_AUTO, expectedColumnNumber);
+ }
+
+ public DFSSingleFileTableReader(String filePath, String delim, int expectedColumnNumber) throws IOException {
+ filePath = HadoopUtil.fixWindowsPath(filePath);
+ this.filePath = filePath;
+ this.delim = delim;
+ this.expectedColumnNumber = expectedColumnNumber;
+
+ FileSystem fs = HadoopUtil.getFileSystem(filePath);
+
+ try {
+ this.reader = new SeqRowReader(HadoopUtil.getCurrentConfiguration(), fs, filePath);
+
+ } catch (IOException e) {
+ if (isExceptionSayingNotSeqFile(e) == false)
+ throw e;
+
+ this.reader = new CsvRowReader(fs, filePath);
+ }
+ }
+
+ private boolean isExceptionSayingNotSeqFile(IOException e) {
+ if (e.getMessage() != null && e.getMessage().contains("not a SequenceFile"))
+ return true;
+
+ if (e instanceof EOFException) // in case the file is very very small
+ return true;
+
+ return false;
+ }
+
+ @Override
+ public boolean next() throws IOException {
+ curLine = reader.nextLine();
+ curColumns = null;
+ return curLine != null;
+ }
+
+ public String getLine() {
+ return curLine;
+ }
+
+ @Override
+ public String[] getRow() {
+ if (curColumns == null) {
+ if (DFSFileTable.DELIM_AUTO.equals(delim))
+ delim = autoDetectDelim(curLine);
+
+ if (delim == null)
+ curColumns = new String[] { curLine };
+ else
+ curColumns = split(curLine, delim);
+ }
+ return curColumns;
+ }
+
+ private String[] split(String line, String delim) {
+ // FIXME CVS line should be parsed considering escapes
+ String[] str = StringSplitter.split(line, delim);
+
+ // un-escape CSV
+ if (DFSFileTable.DELIM_COMMA.equals(delim)) {
+ for (int i = 0; i < str.length; i++) {
+ str[i] = unescapeCsv(str[i]);
+ }
+ }
+
+ return str;
+ }
+
+ private String unescapeCsv(String str) {
+ if (str == null || str.length() < 2)
+ return str;
+
+ str = StringEscapeUtils.unescapeCsv(str);
+
+ // unescapeCsv may not remove the outer most quotes
+ if (str.charAt(0) == CSV_QUOTE && str.charAt(str.length() - 1) == CSV_QUOTE)
+ str = str.substring(1, str.length() - 1);
+
+ return str;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (reader != null)
+ reader.close();
+ }
+
+ private String autoDetectDelim(String line) {
+ if (expectedColumnNumber > 0) {
+ for (String delim : DETECT_DELIMS) {
+ if (StringSplitter.split(line, delim).length == expectedColumnNumber) {
+ logger.info("Auto detect delim to be '" + delim + "', split line to " + expectedColumnNumber + " columns -- " + line);
+ return delim;
+ }
+ }
+ }
+
+ logger.info("Auto detect delim to be null, will take THE-WHOLE-LINE as a single value, for " + filePath);
+ return null;
+ }
+
+ // ============================================================================
+
+ private interface RowReader extends Closeable {
+ String nextLine() throws IOException; // return null on EOF
+ }
+
+ private class SeqRowReader implements RowReader {
+ Reader reader;
+ Writable key;
+ Text value;
+
+ SeqRowReader(Configuration hconf, FileSystem fs, String path) throws IOException {
+ reader = new Reader(hconf, Reader.file(new Path(path)));
+ key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), hconf);
+ value = new Text();
+ }
+
+ @Override
+ public String nextLine() throws IOException {
+ boolean hasNext = reader.next(key, value);
+ if (hasNext)
+ return Bytes.toString(value.getBytes(), 0, value.getLength());
+ else
+ return null;
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+ }
+
+ private class CsvRowReader implements RowReader {
+ BufferedReader reader;
+
+ CsvRowReader(FileSystem fs, String path) throws IOException {
+ FSDataInputStream in = fs.open(new Path(path));
+ reader = new BufferedReader(new InputStreamReader(in, "UTF-8"));
+ }
+
+ @Override
+ public String nextLine() throws IOException {
+ return reader.readLine();
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/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 9bb867b..47eb9c3 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
@@ -27,10 +27,10 @@ import org.apache.kylin.engine.mr.common.HadoopShellExecutable;
import org.apache.kylin.engine.mr.common.MapReduceExecutable;
import org.apache.kylin.engine.mr.steps.CreateDictionaryJob;
import org.apache.kylin.engine.mr.steps.CubingExecutableUtil;
+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.engine.mr.steps.fdc2.FactDistinctColumnsJob2;
import org.apache.kylin.job.constant.ExecutableConstants;
import org.apache.kylin.job.engine.JobEngineConfig;
@@ -63,7 +63,7 @@ public class JobBuilderSupport {
private MapReduceExecutable createFactDistinctColumnsStep(String jobId, boolean withStats) {
MapReduceExecutable result = new MapReduceExecutable();
result.setName(ExecutableConstants.STEP_NAME_FACT_DISTINCT_COLUMNS);
- result.setMapReduceJobClass(FactDistinctColumnsJob2.class);
+ result.setMapReduceJobClass(FactDistinctColumnsJob.class);
StringBuilder cmd = new StringBuilder();
appendMapReduceParameters(cmd);
appendExecCmdParameters(cmd, BatchConstants.ARG_CUBE_NAME, seg.getRealization().getName());
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/SortedColumn.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/SortedColumn.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/SortedColumn.java
new file mode 100644
index 0000000..46b13f4
--- /dev/null
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/SortedColumn.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 org.apache.kylin.dict.ByteComparator;
+import org.apache.kylin.dict.StringBytesConverter;
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.source.ReadableTable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Comparator;
+
+/**
+ * Created by xiefan on 16-11-22.
+ *
+ * Read values from multi col files and ensure their order using a K-Way merge algorithm
+ *
+ * You need to ensure that values inside each file is sorted
+ */
+public class SortedColumn implements ReadableTable {
+
+ private DFSFileTable dfsFileTable;
+
+ private String path;
+
+ private DataType dataType;
+
+ private static final Logger logger = LoggerFactory.getLogger(SortedColumn.class);
+
+ public SortedColumn(String path, DataType dataType) {
+ this.dfsFileTable = new DFSFileTable(path, -1);
+ this.dataType = dataType;
+ }
+
+ @Override
+ public TableReader getReader() throws IOException {
+ final Comparator<String> comparator = getComparatorByType(dataType);
+ return new SortedColumnReader(dfsFileTable.getReaders(), comparator);
+ }
+
+ @Override
+ public TableSignature getSignature() throws IOException {
+ return dfsFileTable.getSignature();
+ }
+
+ private Comparator<String> getComparatorByType(DataType type) {
+ Comparator<String> comparator;
+ if (!type.isNumberFamily()) {
+ comparator = new ByteComparator<>(new StringBytesConverter());
+ } else if (type.isIntegerFamily()) {
+ comparator = new Comparator<String>() {
+ @Override
+ public int compare(String str1, String str2) {
+ try {
+ Long num1 = Long.parseLong(str1);
+ Long num2 = Long.parseLong(str2);
+ return num1.compareTo(num2);
+ } catch (NumberFormatException e) {
+ logger.error("NumberFormatException when parse integer family number.str1:" + str1 + " str2:" + str2);
+ e.printStackTrace();
+ return 0;
+ }
+ }
+ };
+ } else {
+ comparator = new Comparator<String>() {
+ @Override
+ public int compare(String str1, String str2) {
+ try {
+ Double num1 = Double.parseDouble(str1);
+ Double num2 = Double.parseDouble(str2);
+ return num1.compareTo(num2);
+ } catch (NumberFormatException e) {
+ logger.error("NumberFormatException when parse doul family number.str1:" + str1 + " str2:" + str2);
+ return 0;
+ }
+ }
+ };
+ }
+ return comparator;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/SortedColumnReader.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/SortedColumnReader.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/SortedColumnReader.java
new file mode 100644
index 0000000..215198f
--- /dev/null
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/SortedColumnReader.java
@@ -0,0 +1,136 @@
+/*
+ * 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.source.ReadableTable;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.PriorityQueue;
+
+/**
+ * Created by xiefan on 16-11-22.
+ */
+public class SortedColumnReader implements ReadableTable.TableReader {
+ private Collection<ReadableTable.TableReader> readers;
+
+ private Comparator<String> comparator;
+
+ private PriorityQueue<ReaderBuffer> pq;
+
+ private String[] row;
+
+ public SortedColumnReader(Collection<ReadableTable.TableReader> readers, final Comparator<String> comparator) {
+ this.readers = readers;
+ this.comparator = comparator;
+ pq = new PriorityQueue<ReaderBuffer>(11, new Comparator<ReaderBuffer>() {
+ @Override
+ public int compare(ReaderBuffer i, ReaderBuffer j) {
+ boolean isEmpty1 = i.empty();
+ boolean isEmpty2 = j.empty();
+ if (isEmpty1 && isEmpty2)
+ return 0;
+ if (isEmpty1 && !isEmpty2)
+ return 1;
+ if (!isEmpty1 && isEmpty2)
+ return -1;
+ return comparator.compare(i.peek()[0], j.peek()[0]);
+ }
+ });
+ for (ReadableTable.TableReader reader : readers) {
+ if (reader != null) {
+ try {
+ pq.add(new ReaderBuffer(reader));
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+
+ @Override
+ public boolean next() throws IOException {
+ while (pq.size() > 0) {
+ ReaderBuffer buffer = pq.poll();
+ String[] minEntry = buffer.pop();
+ this.row = minEntry;
+ if (buffer.empty()) {
+ pq.remove(buffer);
+ } else {
+ pq.add(buffer); // add it back
+ }
+ if (this.row == null) { //avoid the case of empty file
+ return false;
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public String[] getRow() {
+ return this.row;
+ }
+
+ @Override
+ public void close() throws IOException {
+ for (ReadableTable.TableReader reader : readers)
+ reader.close();
+ }
+
+ static class ReaderBuffer {
+ public ReaderBuffer(ReadableTable.TableReader reader) throws IOException {
+ this.reader = reader;
+ reload();
+ }
+
+ public void close() throws IOException {
+ if (this.reader != null)
+ reader.close();
+ }
+
+ public boolean empty() {
+ return (this.row == null);
+ }
+
+ public String[] peek() {
+ return this.row;
+ }
+
+ public String[] pop() throws IOException {
+ String[] result = this.row;
+ reload();
+ return result;
+ }
+
+ private void reload() throws IOException {
+ if (reader.next()) {
+ row = reader.getRow();
+ } else {
+ this.row = null;
+ }
+ }
+
+ private ReadableTable.TableReader reader;
+
+ private String[] row;
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CreateDictionaryJob.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CreateDictionaryJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CreateDictionaryJob.java
index 5db7c88..7447133 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CreateDictionaryJob.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CreateDictionaryJob.java
@@ -23,7 +23,7 @@ import org.apache.hadoop.util.ToolRunner;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.cube.cli.DictionaryGeneratorCLI;
import org.apache.kylin.dict.DistinctColumnValuesProvider;
-import org.apache.kylin.engine.mr.DFSFileTable;
+import org.apache.kylin.engine.mr.SortedColumn;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.source.ReadableTable;
@@ -54,7 +54,8 @@ public class CreateDictionaryJob extends AbstractHadoopJob {
DictionaryGeneratorCLI.processSegment(config, cubeName, segmentID, new DistinctColumnValuesProvider() {
@Override
public ReadableTable getDistinctValuesFor(TblColRef col) {
- return new DFSFileTable(factColumnsInputPath + "/" + col.getName(), -1);
+ //return new DFSFileTable(factColumnsInputPath + "/" + col.getName(), -1);
+ return new SortedColumn(factColumnsInputPath + "/" + col.getName(), col.getType());
}
});
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnPartitioner.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnPartitioner.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnPartitioner.java
index 1073a46..e8817a5 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnPartitioner.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnPartitioner.java
@@ -24,10 +24,11 @@ import org.apache.kylin.common.util.BytesUtil;
/**
*/
-public class FactDistinctColumnPartitioner extends Partitioner<Text, Text> {
+public class FactDistinctColumnPartitioner extends Partitioner<SelfDefineSortableKey, Text> {
@Override
- public int getPartition(Text key, Text value, int numReduceTasks) {
+ public int getPartition(SelfDefineSortableKey skey, Text value, int numReduceTasks) {
+ Text key = skey.getText();
if (key.getBytes()[0] == FactDistinctHiveColumnsMapper.MARK_FOR_HLL) {
// the last reducer is for merging hll
return numReduceTasks - 1;
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsCombiner.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsCombiner.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsCombiner.java
index 2dda047..a367bc6 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsCombiner.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsCombiner.java
@@ -26,7 +26,7 @@ import org.apache.kylin.engine.mr.KylinReducer;
/**
* @author yangli9
*/
-public class FactDistinctColumnsCombiner extends KylinReducer<Text, Text, Text, Text> {
+public class FactDistinctColumnsCombiner extends KylinReducer<SelfDefineSortableKey, Text, SelfDefineSortableKey, Text> {
@Override
protected void setup(Context context) throws IOException {
@@ -34,7 +34,7 @@ public class FactDistinctColumnsCombiner extends KylinReducer<Text, Text, Text,
}
@Override
- public void doReduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
+ public void doReduce(SelfDefineSortableKey key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
// for hll, each key only has one output, no need to do local combine;
// for normal col, values are empty text
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsJob.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsJob.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsJob.java
index 551ce33..863ab8f 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsJob.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsJob.java
@@ -133,7 +133,7 @@ public class FactDistinctColumnsJob extends AbstractHadoopJob {
job.setMapperClass(FactDistinctHiveColumnsMapper.class);
job.setCombinerClass(FactDistinctColumnsCombiner.class);
- job.setMapOutputKeyClass(Text.class);
+ job.setMapOutputKeyClass(SelfDefineSortableKey.class);
job.setMapOutputValueClass(Text.class);
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapperBase.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapperBase.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapperBase.java
index 284242d..74aebb0 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapperBase.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapperBase.java
@@ -43,7 +43,7 @@ import org.apache.kylin.metadata.model.TblColRef;
/**
*/
-abstract public class FactDistinctColumnsMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VALUEIN, Text, Text> {
+abstract public class FactDistinctColumnsMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VALUEIN, SelfDefineSortableKey, Text> {
protected String cubeName;
protected CubeInstance cube;
@@ -54,6 +54,7 @@ abstract public class FactDistinctColumnsMapperBase<KEYIN, VALUEIN> extends Kyli
protected IMRTableInputFormat flatTableInputFormat;
protected Text outputKey = new Text();
+ protected SelfDefineSortableKey sortableKey = new SelfDefineSortableKey();
protected Text outputValue = new Text();
protected int errorRecordCounter = 0;
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
index 9bbb151..9722b9c 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
@@ -53,7 +53,7 @@ import com.google.common.collect.Maps;
/**
*/
-public class FactDistinctColumnsReducer extends KylinReducer<Text, Text, NullWritable, Text> {
+public class FactDistinctColumnsReducer extends KylinReducer<SelfDefineSortableKey, Text, NullWritable, Text> {
private List<TblColRef> columnList;
private String statisticsOutput = null;
@@ -130,8 +130,8 @@ public class FactDistinctColumnsReducer extends KylinReducer<Text, Text, NullWri
}
@Override
- public void doReduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
-
+ public void doReduce(SelfDefineSortableKey skey, Iterable<Text> values, Context context) throws IOException, InterruptedException {
+ Text key = skey.getText();
if (isStatistics == true) {
// for hll
long cuboidId = Bytes.toLong(key.getBytes(), 1, Bytes.SIZEOF_LONG);
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctHiveColumnsMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctHiveColumnsMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctHiveColumnsMapper.java
index c3a3f9b..7776172 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctHiveColumnsMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctHiveColumnsMapper.java
@@ -35,6 +35,7 @@ import com.google.common.collect.Lists;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
+import org.apache.kylin.metadata.datatype.DataType;
import org.apache.kylin.metadata.model.TblColRef;
/**
@@ -150,11 +151,20 @@ public class FactDistinctHiveColumnsMapper<KEYIN> extends FactDistinctColumnsMap
//for the uhc
reducerIndex = columnIndexToReducerBeginId.get(i) + (fieldValue.hashCode() & 0x7fffffff) % uhcReducerCount;
}
-
keyBuffer.put(Bytes.toBytes(reducerIndex)[3]);
keyBuffer.put(Bytes.toBytes(fieldValue));
outputKey.set(keyBuffer.array(), offset, keyBuffer.position() - offset);
- context.write(outputKey, EMPTY_TEXT);
+ sortableKey.setText(outputKey);
+ //judge type
+ DataType type = factDictCols.get(i).getType();
+ if (!type.isNumberFamily()) {
+ sortableKey.setTypeId((byte) TypeFlag.NONE_NUMERIC_TYPE.ordinal());
+ } else if (type.isIntegerFamily()) {
+ sortableKey.setTypeId((byte) TypeFlag.INTEGER_FAMILY_TYPE.ordinal());
+ } else {
+ sortableKey.setTypeId((byte) TypeFlag.DOUBLE_FAMILY_TYPE.ordinal());
+ }
+ context.write(sortableKey, EMPTY_TEXT);
}
} catch (Exception ex) {
handleErrorRecord(row, ex);
@@ -172,7 +182,9 @@ public class FactDistinctHiveColumnsMapper<KEYIN> extends FactDistinctColumnsMap
keyBuffer.put(MARK_FOR_PARTITION_COL);
keyBuffer.put(Bytes.toBytes(fieldValue));
outputKey.set(keyBuffer.array(), offset, keyBuffer.position() - offset);
- context.write(outputKey, EMPTY_TEXT);
+ sortableKey.setText(outputKey);
+ sortableKey.setTypeId((byte) 0);
+ context.write(sortableKey, EMPTY_TEXT);
}
}
}
@@ -221,7 +233,9 @@ public class FactDistinctHiveColumnsMapper<KEYIN> extends FactDistinctColumnsMap
hllBuf.clear();
hll.writeRegisters(hllBuf);
outputValue.set(hllBuf.array(), 0, hllBuf.position());
- context.write(outputKey, outputValue);
+ sortableKey.setText(outputKey);
+ sortableKey.setTypeId((byte) 0);
+ context.write(sortableKey, outputValue);
}
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SelfDefineSortableKey.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SelfDefineSortableKey.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SelfDefineSortableKey.java
new file mode 100644
index 0000000..4ca3a90
--- /dev/null
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SelfDefineSortableKey.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+package org.apache.kylin.engine.mr.steps;
+
+
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableComparable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+
+/**
+ * Created by xiefan on 16-11-1.
+ */
+public class SelfDefineSortableKey implements WritableComparable<SelfDefineSortableKey> {
+
+ private byte typeId; //non-numeric(0000 0000) int(0000 0001) other numberic(0000 0010)
+
+ private Text text;
+
+ private static final Logger logger = LoggerFactory.getLogger(SelfDefineSortableKey.class);
+
+ public SelfDefineSortableKey() {
+ }
+
+ public SelfDefineSortableKey(byte typeId, Text text) {
+ this.typeId = typeId;
+ this.text = text;
+ }
+
+ @Override
+ public int compareTo(SelfDefineSortableKey o) {
+ if (!o.isNumberFamily()) {
+ return this.text.compareTo(o.text);
+ } else {
+ byte[] data1 = this.text.getBytes();
+ byte[] data2 = o.text.getBytes();
+ String str1 = new String(data1, 1, data1.length - 1);
+ String str2 = new String(data2, 1, data2.length - 1);
+ if (str1 == null || str1.equals("") || str2 == null || str2.equals("")) {
+ //should not achieve here
+ logger.error("none numeric value!");
+ return 0;
+ }
+ if (o.isIntegerFamily()) { //integer type
+ try {
+ Long num1 = Long.parseLong(str1);
+ Long num2 = Long.parseLong(str2);
+ return num1.compareTo(num2);
+ } catch (NumberFormatException e) {
+ logger.error("NumberFormatException when parse integer family number.str1:" + str1 + " str2:" + str2);
+ e.printStackTrace();
+ return 0;
+ }
+ } else { //other numeric type
+ try {
+ Double num1 = Double.parseDouble(str1);
+ Double num2 = Double.parseDouble(str2);
+ return num1.compareTo(num2);
+ } catch (NumberFormatException e) {
+ logger.error("NumberFormatException when parse doul family number.str1:" + str1 + " str2:" + str2);
+ return 0;
+ }
+ }
+ }
+ }
+
+ @Override
+ public void write(DataOutput dataOutput) throws IOException {
+ dataOutput.writeByte(typeId);
+ text.write(dataOutput);
+ }
+
+ @Override
+ public void readFields(DataInput dataInput) throws IOException {
+ this.typeId = dataInput.readByte();
+ this.text = new Text();
+ text.readFields(dataInput);
+ }
+
+ public short getTypeId() {
+ return typeId;
+ }
+
+ public Text getText() {
+ return text;
+ }
+
+ public boolean isNumberFamily() {
+ if (typeId == TypeFlag.NONE_NUMERIC_TYPE.ordinal()) return false;
+ return true;
+ }
+
+ public boolean isIntegerFamily() {
+ return (typeId == TypeFlag.INTEGER_FAMILY_TYPE.ordinal());
+ }
+
+ public boolean isOtherNumericFamily() {
+ return (typeId == TypeFlag.DOUBLE_FAMILY_TYPE.ordinal());
+ }
+
+ public void setTypeId(byte typeId) {
+ this.typeId = typeId;
+ }
+
+ public void setText(Text text) {
+ this.text = text;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/TypeFlag.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/TypeFlag.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/TypeFlag.java
new file mode 100644
index 0000000..3279106
--- /dev/null
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/TypeFlag.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.kylin.engine.mr.steps;
+
+/**
+ * Created by xiefan on 16-11-2.
+ */
+public enum TypeFlag {
+ NONE_NUMERIC_TYPE,
+ INTEGER_FAMILY_TYPE,
+ DOUBLE_FAMILY_TYPE
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/UpdateCubeInfoAfterBuildStep.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/UpdateCubeInfoAfterBuildStep.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/UpdateCubeInfoAfterBuildStep.java
index 4d71f5d..6e18f9b 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/UpdateCubeInfoAfterBuildStep.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/UpdateCubeInfoAfterBuildStep.java
@@ -29,7 +29,7 @@ import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.engine.mr.CubingJob;
-import org.apache.kylin.engine.mr.DFSFileTable;
+import org.apache.kylin.engine.mr.SortedColumn;
import org.apache.kylin.engine.mr.common.BatchConstants;
import org.apache.kylin.job.exception.ExecuteException;
import org.apache.kylin.job.execution.AbstractExecutable;
@@ -100,7 +100,8 @@ public class UpdateCubeInfoAfterBuildStep extends AbstractExecutable {
}
final String factDistinctPath = this.getParams().get(BatchConstants.CFG_OUTPUT_PATH);
- final ReadableTable readableTable = new DFSFileTable(factDistinctPath + "/" + partitionCol.getName(), -1);
+ //final ReadableTable readableTable = new DFSFileTable(factDistinctPath + "/" + partitionCol.getName(), -1);
+ final ReadableTable readableTable = new SortedColumn(factDistinctPath + "/" + partitionCol.getName(), partitionCol.getType());
final ReadableTable.TableReader tableReader = readableTable.getReader();
long minValue = Long.MAX_VALUE, maxValue = Long.MIN_VALUE;
try {
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnPartitioner2.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnPartitioner2.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnPartitioner2.java
deleted file mode 100644
index dfc6b2c..0000000
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnPartitioner2.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kylin.engine.mr.steps.fdc2;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapreduce.Partitioner;
-import org.apache.kylin.common.util.BytesUtil;
-
-/**
- */
-public class FactDistinctColumnPartitioner2 extends Partitioner<SelfDefineSortableKey, Text> {
- private Configuration conf;
-
- @Override
- public int getPartition(SelfDefineSortableKey key, Text value, int numReduceTasks) {
-
- if (key.getText().getBytes()[0] == FactDistinctHiveColumnsMapper2.MARK_FOR_HLL) {
- // the last reducer is for merging hll
- return numReduceTasks - 1;
- } else if (key.getText().getBytes()[0] == FactDistinctHiveColumnsMapper2.MARK_FOR_PARTITION_COL) {
- // the last reducer is for merging hll
- return numReduceTasks - 2;
- } else {
- int colIndex = BytesUtil.readUnsigned(key.getText().getBytes(), 0, 1);
- return colIndex;
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsCombiner2.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsCombiner2.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsCombiner2.java
deleted file mode 100644
index 289edd0..0000000
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsCombiner2.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.steps.fdc2;
-
-import org.apache.hadoop.io.Text;
-import org.apache.kylin.engine.mr.KylinReducer;
-
-import java.io.IOException;
-
-/**
- * @author yangli9
- */
-public class FactDistinctColumnsCombiner2 extends KylinReducer<SelfDefineSortableKey, Text, SelfDefineSortableKey, Text> {
-
- @Override
- protected void setup(Context context) throws IOException {
- super.bindCurrentConfiguration(context.getConfiguration());
- }
-
- @Override
- public void doReduce(SelfDefineSortableKey key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
-
-
- // for hll, each key only has one output, no need to do local combine;
- // for normal col, values are empty text
- context.write(key, values.iterator().next());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsJob2.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsJob2.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsJob2.java
deleted file mode 100644
index 2e84f45..0000000
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsJob2.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
-*/
-
-package org.apache.kylin.engine.mr.steps.fdc2;
-
-import org.apache.commons.cli.Options;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
-import org.apache.hadoop.util.ToolRunner;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.cube.CubeInstance;
-import org.apache.kylin.cube.CubeManager;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.engine.mr.IMRInput.IMRTableInputFormat;
-import org.apache.kylin.engine.mr.MRUtil;
-import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
-import org.apache.kylin.engine.mr.common.BatchConstants;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.List;
-
-/**
- */
-public class FactDistinctColumnsJob2 extends AbstractHadoopJob {
- protected static final Logger logger = LoggerFactory.getLogger(FactDistinctColumnsJob2.class);
-
- @Override
- public int run(String[] args) throws Exception {
- Options options = new Options();
-
- try {
- options.addOption(OPTION_JOB_NAME);
- options.addOption(OPTION_CUBE_NAME);
- options.addOption(OPTION_CUBING_JOB_ID);
- options.addOption(OPTION_OUTPUT_PATH);
- options.addOption(OPTION_SEGMENT_ID);
- options.addOption(OPTION_STATISTICS_ENABLED);
- options.addOption(OPTION_STATISTICS_OUTPUT);
- options.addOption(OPTION_STATISTICS_SAMPLING_PERCENT);
- parseOptions(options, args);
-
- job = Job.getInstance(getConf(), getOptionValue(OPTION_JOB_NAME));
- String job_id = getOptionValue(OPTION_CUBING_JOB_ID);
- job.getConfiguration().set(BatchConstants.ARG_CUBING_JOB_ID, job_id);
- String cubeName = getOptionValue(OPTION_CUBE_NAME);
- Path output = new Path(getOptionValue(OPTION_OUTPUT_PATH));
-
- String segmentID = getOptionValue(OPTION_SEGMENT_ID);
- String statistics_enabled = getOptionValue(OPTION_STATISTICS_ENABLED);
- String statistics_output = getOptionValue(OPTION_STATISTICS_OUTPUT);
- String statistics_sampling_percent = getOptionValue(OPTION_STATISTICS_SAMPLING_PERCENT);
-
- // ----------------------------------------------------------------------------
- // add metadata to distributed cache
- CubeManager cubeMgr = CubeManager.getInstance(KylinConfig.getInstanceFromEnv());
- CubeInstance cube = cubeMgr.getCube(cubeName);
- List<TblColRef> columnsNeedDict = cubeMgr.getAllDictColumnsOnFact(cube.getDescriptor());
-
- job.getConfiguration().set(BatchConstants.CFG_CUBE_NAME, cubeName);
- job.getConfiguration().set(BatchConstants.CFG_CUBE_SEGMENT_ID, segmentID);
- job.getConfiguration().set(BatchConstants.CFG_STATISTICS_ENABLED, statistics_enabled);
- job.getConfiguration().set(BatchConstants.CFG_STATISTICS_OUTPUT, statistics_output);
- job.getConfiguration().set(BatchConstants.CFG_STATISTICS_SAMPLING_PERCENT, statistics_sampling_percent);
- logger.info("Starting: " + job.getJobName());
- logger.info("using FactDistinctColumnsJob2");
-
- setJobClasspath(job, cube.getConfig());
-
- CubeSegment segment = cube.getSegmentById(segmentID);
- if (segment == null) {
- logger.error("Failed to find {} in cube {}", segmentID, cube);
- System.out.println("Failed to find {} in cube {} " + segmentID + "," + cube);
- for (CubeSegment s : cube.getSegments()) {
- logger.error(s.getName() + " with status " + s.getStatus());
- System.out.println(s.getName() + " with status " + s.getStatus());
- }
- throw new IllegalStateException();
- } else {
- logger.info("Found segment: " + segment);
- System.out.println("Found segment " + segment);
- }
- setupMapper(cube.getSegmentById(segmentID));
- setupReducer(output, "true".equalsIgnoreCase(statistics_enabled) ? columnsNeedDict.size() + 2 : columnsNeedDict.size());
-
- attachKylinPropsAndMetadata(cube, job.getConfiguration());
-
- return waitForCompletion(job);
-
- } finally {
- if (job != null)
- cleanupTempConfFile(job.getConfiguration());
- }
-
- }
-
- private void setupMapper(CubeSegment cubeSeg) throws IOException {
- IMRTableInputFormat flatTableInputFormat = MRUtil.getBatchCubingInputSide(cubeSeg).getFlatTableInputFormat();
- flatTableInputFormat.configureJob(job);
-
- job.setMapperClass(FactDistinctHiveColumnsMapper2.class);
- job.setCombinerClass(FactDistinctColumnsCombiner2.class);
- job.setMapOutputKeyClass(SelfDefineSortableKey.class);
- job.setMapOutputValueClass(Text.class);
- }
-
- private void setupReducer(Path output, int numberOfReducers) throws IOException {
- job.setReducerClass(FactDistinctColumnsReducer2.class);
- job.setOutputFormatClass(SequenceFileOutputFormat.class);
- job.setOutputKeyClass(NullWritable.class);
- job.setOutputValueClass(Text.class);
- job.setPartitionerClass(FactDistinctColumnPartitioner2.class);
- job.setNumReduceTasks(numberOfReducers);
-
- FileOutputFormat.setOutputPath(job, output);
- job.getConfiguration().set(BatchConstants.CFG_OUTPUT_PATH, output.toString());
-
- deletePath(job.getConfiguration(), output);
- }
-
- public static void main(String[] args) throws Exception {
- FactDistinctColumnsJob2 job = new FactDistinctColumnsJob2();
- int exitCode = ToolRunner.run(job, args);
- System.exit(exitCode);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2b7fe610/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsMapperBase2.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsMapperBase2.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsMapperBase2.java
deleted file mode 100644
index 037afeb..0000000
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/fdc2/FactDistinctColumnsMapperBase2.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kylin.engine.mr.steps.fdc2;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.Text;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.cube.CubeInstance;
-import org.apache.kylin.cube.CubeManager;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.cuboid.Cuboid;
-import org.apache.kylin.cube.model.CubeDesc;
-import org.apache.kylin.cube.model.CubeJoinedFlatTableEnrich;
-import org.apache.kylin.engine.EngineFactory;
-import org.apache.kylin.engine.mr.IMRInput.IMRTableInputFormat;
-import org.apache.kylin.engine.mr.KylinMapper;
-import org.apache.kylin.engine.mr.MRUtil;
-import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
-import org.apache.kylin.engine.mr.common.BatchConstants;
-import org.apache.kylin.metadata.model.TblColRef;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.List;
-
-/**
- */
-
-abstract public class FactDistinctColumnsMapperBase2<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VALUEIN, SelfDefineSortableKey, Text> {
-
- protected String cubeName;
- protected CubeInstance cube;
- protected CubeSegment cubeSeg;
- protected CubeDesc cubeDesc;
- protected long baseCuboidId;
- protected List<TblColRef> factDictCols;
- protected IMRTableInputFormat flatTableInputFormat;
-
- protected Text outputKey = new Text();
- protected SelfDefineSortableKey sortableKey = new SelfDefineSortableKey();
- protected Text outputValue = new Text();
- protected int errorRecordCounter = 0;
-
- protected CubeJoinedFlatTableEnrich intermediateTableDesc;
- protected int[] dictionaryColumnIndex;
-
- @Override
- protected void setup(Context context) throws IOException {
- Configuration conf = context.getConfiguration();
- bindCurrentConfiguration(conf);
- KylinConfig config = AbstractHadoopJob.loadKylinPropsAndMetadata();
-
- cubeName = conf.get(BatchConstants.CFG_CUBE_NAME);
- cube = CubeManager.getInstance(config).getCube(cubeName);
- cubeSeg = cube.getSegmentById(conf.get(BatchConstants.CFG_CUBE_SEGMENT_ID));
- cubeDesc = cube.getDescriptor();
- baseCuboidId = Cuboid.getBaseCuboidId(cubeDesc);
- factDictCols = CubeManager.getInstance(config).getAllDictColumnsOnFact(cubeDesc);
-
- flatTableInputFormat = MRUtil.getBatchCubingInputSide(cubeSeg).getFlatTableInputFormat();
-
- intermediateTableDesc = new CubeJoinedFlatTableEnrich(EngineFactory.getJoinedFlatTableDesc(cubeSeg), cubeDesc);
- dictionaryColumnIndex = new int[factDictCols.size()];
- for (int i = 0; i < factDictCols.size(); i++) {
- TblColRef colRef = factDictCols.get(i);
- int columnIndexOnFlatTbl = intermediateTableDesc.getColumnIndex(colRef);
- dictionaryColumnIndex[i] = columnIndexOnFlatTbl;
- }
-
- }
-
- protected void handleErrorRecord(String[] record, Exception ex) throws IOException {
-
- System.err.println("Insane record: " + Arrays.toString(record));
- ex.printStackTrace(System.err);
-
- errorRecordCounter++;
- if (errorRecordCounter > BatchConstants.ERROR_RECORD_LOG_THRESHOLD) {
- if (ex instanceof IOException)
- throw (IOException) ex;
- else if (ex instanceof RuntimeException)
- throw (RuntimeException) ex;
- else
- throw new RuntimeException("", ex);
- }
- }
-}