You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by sh...@apache.org on 2015/09/29 03:03:53 UTC
[02/27] incubator-kylin git commit: KYLIN-943 TopN in cube builder,
on the way
KYLIN-943 TopN in cube builder, on the way
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/db5952cf
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/db5952cf
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/db5952cf
Branch: refs/heads/2.x-staging
Commit: db5952cfcb7f4eb2ed90a328489b1a1612b035f8
Parents: 76b7f03
Author: shaofengshi <sh...@apache.org>
Authored: Tue Sep 1 16:37:46 2015 +0800
Committer: shaofengshi <sh...@apache.org>
Committed: Tue Sep 29 08:56:33 2015 +0800
----------------------------------------------------------------------
.../apache/kylin/common/topn/TopNCounter.java | 52 +++++++
.../java/org/apache/kylin/cube/CubeManager.java | 4 +-
.../kylin/cube/cli/DictionaryGeneratorCLI.java | 18 +--
.../org/apache/kylin/cube/model/CubeDesc.java | 35 ++++-
.../cube/model/CubeJoinedFlatTableDesc.java | 23 ++-
.../kylin/metadata/measure/TopNAggregator.java | 1 -
.../measure/serializer/DataTypeSerializer.java | 4 +
.../serializer/TopNCounterSerializer.java | 64 +++++++-
.../kylin/metadata/model/ParameterDesc.java | 18 +--
.../engine/mr/steps/BaseCuboidMapperBase.java | 79 +++++++---
.../mr/steps/FactDistinctColumnsMapperBase.java | 38 +++--
.../mr/steps/FactDistinctColumnsReducer.java | 4 +-
.../mr/steps/FactDistinctHiveColumnsMapper.java | 14 +-
.../localmeta/cube/test_kylin_cube_topn.json | 10 ++
.../cube_desc/test_kylin_cube_topn_desc.json | 146 +++++++++++++++++++
15 files changed, 429 insertions(+), 81 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/core-common/src/main/java/org/apache/kylin/common/topn/TopNCounter.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/topn/TopNCounter.java b/core-common/src/main/java/org/apache/kylin/common/topn/TopNCounter.java
index 2f337c2..51fe1b2 100644
--- a/core-common/src/main/java/org/apache/kylin/common/topn/TopNCounter.java
+++ b/core-common/src/main/java/org/apache/kylin/common/topn/TopNCounter.java
@@ -18,9 +18,11 @@
package org.apache.kylin.common.topn;
+import com.google.common.collect.Lists;
import org.apache.kylin.common.util.Pair;
import java.io.*;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -37,6 +39,8 @@ import java.util.Map;
* @param <T> type of data in the stream to be summarized
*/
public class TopNCounter<T> implements ITopK<T>, Externalizable {
+
+ public static final int EXTRA_SPACE_RATE = 50;
protected class Bucket {
@@ -259,6 +263,26 @@ public class TopNCounter<T> implements ITopK<T>, Externalizable {
}
}
+ public void fromExternal(int size, double[] counters, List<T> items) {
+ this.bucketList = new DoublyLinkedList<Bucket>();
+
+ this.counterMap = new HashMap<T, ListNode2<Counter<T>>>(size);
+
+ Bucket currentBucket = null;
+ ListNode2<Bucket> currentBucketNode = null;
+ for (int i = 0; i < size; i++) {
+ Counter<T> c = new Counter<T>();
+ c.count = counters[i];
+ c.item = items.get(i);
+ if (currentBucket == null || c.count != currentBucket.count) {
+ currentBucket = new Bucket(c.count);
+ currentBucketNode = bucketList.add(currentBucket);
+ }
+ c.bucketNode = currentBucketNode;
+ counterMap.put(c.item, currentBucket.counterList.add(c));
+ }
+ }
+
@Override
public void writeExternal(ObjectOutput out) throws IOException {
out.writeInt(this.capacity);
@@ -363,4 +387,32 @@ public class TopNCounter<T> implements ITopK<T>, Externalizable {
}
}
+
+ public double[] getCounters() {
+ double[] counters = new double[size()];
+ int index = 0;
+
+ for (ListNode2<Bucket> bNode = bucketList.head(); bNode != null; bNode = bNode.getPrev()) {
+ Bucket b = bNode.getValue();
+ for (Counter<T> c : b.counterList) {
+ counters[index] = c.count;
+ index ++;
+ }
+ }
+
+ return counters;
+ }
+
+ public List<T> getItems() {
+ List<T> items = Lists.newArrayList();
+ for (ListNode2<Bucket> bNode = bucketList.head(); bNode != null; bNode = bNode.getPrev()) {
+ Bucket b = bNode.getValue();
+ for (Counter<T> c : b.counterList) {
+ items.add(c.item);
+ }
+ }
+
+ return items;
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java b/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
index c6e6b88..2232f01 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
@@ -163,11 +163,11 @@ public class CubeManager implements IRealizationProvider {
public DictionaryInfo buildDictionary(CubeSegment cubeSeg, TblColRef col, DistinctColumnValuesProvider factTableValueProvider) throws IOException {
CubeDesc cubeDesc = cubeSeg.getCubeDesc();
- if (!cubeDesc.getRowkey().isUseDictionary(col))
+ if (!cubeDesc.getAllColumnsNeedDictionary().contains(col))
return null;
DictionaryManager dictMgr = getDictionaryManager();
- DictionaryInfo dictInfo = dictMgr.buildDictionary(cubeDesc.getModel(), cubeDesc.getRowkey().getDictionary(col), col, factTableValueProvider);
+ DictionaryInfo dictInfo = dictMgr.buildDictionary(cubeDesc.getModel(), "true", col, factTableValueProvider);
if (dictInfo != null) {
cubeSeg.putDictResPath(col, dictInfo.getResourcePath());
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/core-cube/src/main/java/org/apache/kylin/cube/cli/DictionaryGeneratorCLI.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/cli/DictionaryGeneratorCLI.java b/core-cube/src/main/java/org/apache/kylin/cube/cli/DictionaryGeneratorCLI.java
index 714b309..0e7a7ac 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/cli/DictionaryGeneratorCLI.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/cli/DictionaryGeneratorCLI.java
@@ -18,8 +18,6 @@
package org.apache.kylin.cube.cli;
-import java.io.IOException;
-
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
@@ -31,6 +29,8 @@ import org.apache.kylin.metadata.model.TblColRef;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
public class DictionaryGeneratorCLI {
private static final Logger logger = LoggerFactory.getLogger(DictionaryGeneratorCLI.class);
@@ -45,15 +45,13 @@ public class DictionaryGeneratorCLI {
private static void processSegment(KylinConfig config, CubeSegment cubeSeg, DistinctColumnValuesProvider factTableValueProvider) throws IOException {
CubeManager cubeMgr = CubeManager.getInstance(config);
- for (DimensionDesc dim : cubeSeg.getCubeDesc().getDimensions()) {
- // dictionary
- for (TblColRef col : dim.getColumnRefs()) {
- if (cubeSeg.getCubeDesc().getRowkey().isUseDictionary(col)) {
- logger.info("Building dictionary for " + col);
- cubeMgr.buildDictionary(cubeSeg, col, factTableValueProvider);
- }
- }
+ // dictionary
+ for (TblColRef col : cubeSeg.getCubeDesc().getAllColumnsNeedDictionary()) {
+ logger.info("Building dictionary for " + col);
+ cubeMgr.buildDictionary(cubeSeg, col, factTableValueProvider);
+ }
+ for (DimensionDesc dim : cubeSeg.getCubeDesc().getDimensions()) {
// build snapshot
if (dim.getTable() != null && !dim.getTable().equalsIgnoreCase(cubeSeg.getCubeDesc().getFactTable())) {
// CubeSegment seg = cube.getTheOnlySegment();
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
index 0060e09..90b5474 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
@@ -129,6 +129,7 @@ public class CubeDesc extends RootPersistentEntity {
private Map<String, Map<String, TblColRef>> columnMap = new HashMap<String, Map<String, TblColRef>>();
private LinkedHashSet<TblColRef> allColumns = new LinkedHashSet<TblColRef>();
private LinkedHashSet<TblColRef> dimensionColumns = new LinkedHashSet<TblColRef>();
+ private LinkedHashSet<TblColRef> measureDisplayColumns = new LinkedHashSet<TblColRef>();
private Map<TblColRef, DeriveInfo> derivedToHostMap = Maps.newHashMap();
private Map<Array<TblColRef>, List<DeriveInfo>> hostToDerivedMap = Maps.newHashMap();
@@ -628,18 +629,28 @@ public class CubeDesc extends RootPersistentEntity {
ParameterDesc p = f.getParameter();
p.normalizeColumnValue();
+ ArrayList<TblColRef> colRefs = Lists.newArrayList();
if (p.isColumnType()) {
- ArrayList<TblColRef> colRefs = Lists.newArrayList();
for (String cName : p.getValue().split("\\s*,\\s*")) {
ColumnDesc sourceColumn = factTable.findColumnByName(cName);
TblColRef colRef = new TblColRef(sourceColumn);
colRefs.add(colRef);
allColumns.add(colRef);
}
- if (colRefs.isEmpty() == false)
- p.setColRefs(colRefs);
}
+ // for topN
+ if (StringUtils.isNotEmpty(p.getDisplayColumn())) {
+ ColumnDesc sourceColumn = factTable.findColumnByName(p.getDisplayColumn());
+ TblColRef colRef = new TblColRef(sourceColumn);
+ colRefs.add(colRef);
+ measureDisplayColumns.add(colRef);
+ allColumns.add(colRef);
+ }
+
+ if (colRefs.isEmpty() == false)
+ p.setColRefs(colRefs);
+
// verify holistic count distinct as a dependent measure
if (m.getFunction().isHolisticCountDistinct() && StringUtils.isBlank(m.getDependentMeasureRef())) {
throw new IllegalStateException(m + " is a holistic count distinct but it has no DependentMeasureRef defined!");
@@ -799,4 +810,22 @@ public class CubeDesc extends RootPersistentEntity {
this.engineType = engineType;
}
+
+ public List<TblColRef> getAllColumnsNeedDictionary() {
+ List<TblColRef> result = Lists.newArrayList();
+
+ for (RowKeyColDesc rowKeyColDesc : rowkey.getRowKeyColumns()) {
+ TblColRef colRef = rowKeyColDesc.getColRef();
+ if (rowkey.isUseDictionary(colRef)) {
+ result.add(colRef);
+ }
+ }
+
+ for (TblColRef colRef : measureDisplayColumns) {
+ if (!result.contains(colRef))
+ result.add(colRef);
+ }
+ return result;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableDesc.java
index 4ac8848..adaf542 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeJoinedFlatTableDesc.java
@@ -30,9 +30,12 @@ public class CubeJoinedFlatTableDesc implements IJoinedFlatTableDesc {
private List<IntermediateColumnDesc> columnList = Lists.newArrayList();
+ private Map<String, Integer> columnIndexMap;
+
public CubeJoinedFlatTableDesc(CubeDesc cubeDesc, CubeSegment cubeSegment) {
this.cubeDesc = cubeDesc;
this.cubeSegment = cubeSegment;
+ this.columnIndexMap = Maps.newHashMap();
parseCubeDesc();
}
@@ -55,26 +58,25 @@ public class CubeJoinedFlatTableDesc implements IJoinedFlatTableDesc {
this.tableName = "kylin_intermediate_" + cubeDesc.getName() + "_" + cubeSegment.getName();
}
- Map<String, Integer> dimensionIndexMap = Maps.newHashMap();
int columnIndex = 0;
for (TblColRef col : cubeDesc.listDimensionColumnsExcludingDerived()) {
- dimensionIndexMap.put(colName(col.getCanonicalName()), columnIndex);
+ columnIndexMap.put(colName(col.getCanonicalName()), columnIndex);
columnList.add(new IntermediateColumnDesc(String.valueOf(columnIndex), col));
columnIndex++;
}
- // build index
+ // build index for rowkey columns
List<TblColRef> cuboidColumns = baseCuboid.getColumns();
rowKeyColumnIndexes = new int[rowkeyColCount];
for (int i = 0; i < rowkeyColCount; i++) {
String colName = colName(cuboidColumns.get(i).getCanonicalName());
- Integer dimIdx = dimensionIndexMap.get(colName);
+ Integer dimIdx = columnIndexMap.get(colName);
if (dimIdx == null) {
throw new RuntimeException("Can't find column " + colName);
}
rowKeyColumnIndexes[i] = dimIdx;
}
-
+
List<MeasureDesc> measures = cubeDesc.getMeasures();
int measureSize = measures.size();
measureColumnIndexes = new int[measureSize][];
@@ -90,6 +92,7 @@ public class CubeJoinedFlatTableDesc implements IJoinedFlatTableDesc {
measureColumnIndexes[i][j] = contains(columnList, c);
if (measureColumnIndexes[i][j] < 0) {
measureColumnIndexes[i][j] = columnIndex;
+ columnIndexMap.put(colName(c.getCanonicalName()), columnIndex);
columnList.add(new IntermediateColumnDesc(String.valueOf(columnIndex), c));
columnIndex++;
}
@@ -154,4 +157,14 @@ public class CubeJoinedFlatTableDesc implements IJoinedFlatTableDesc {
private static String colName(String canonicalColName) {
return canonicalColName.replace(".", "_");
}
+
+ public int getColumnIndex(TblColRef colRef) {
+ String key = colName(colRef.getCanonicalName());
+ Integer index = columnIndexMap.get(key);
+ if (index == null)
+ throw new IllegalArgumentException("Column " + colRef.toString() + " wasn't found on flat table.");
+
+ return index.intValue();
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/core-metadata/src/main/java/org/apache/kylin/metadata/measure/TopNAggregator.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/measure/TopNAggregator.java b/core-metadata/src/main/java/org/apache/kylin/metadata/measure/TopNAggregator.java
index 9cea4cc..0de5fe8 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/measure/TopNAggregator.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/measure/TopNAggregator.java
@@ -53,7 +53,6 @@ public class TopNAggregator extends MeasureAggregator<TopNCounter<ByteArray>> {
@Override
public int getMemBytesEstimate() {
- // 1024 + 60 returned by AggregationCacheMemSizeTest
return 8 * capacity;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/core-metadata/src/main/java/org/apache/kylin/metadata/measure/serializer/DataTypeSerializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/measure/serializer/DataTypeSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/measure/serializer/DataTypeSerializer.java
index e68beda..4fadbb0 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/measure/serializer/DataTypeSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/measure/serializer/DataTypeSerializer.java
@@ -63,6 +63,10 @@ abstract public class DataTypeSerializer<T> implements BytesSerializer<T> {
return new HLLCSerializer(type);
}
+ if (type.isTopN()) {
+ return new TopNCounterSerializer(type);
+ }
+
Class<?> clz = implementations.get(type.getName());
if (clz == null)
throw new RuntimeException("No MeasureSerializer for type " + type);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/core-metadata/src/main/java/org/apache/kylin/metadata/measure/serializer/TopNCounterSerializer.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/measure/serializer/TopNCounterSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/measure/serializer/TopNCounterSerializer.java
index 1c6c442..56ed85d 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/measure/serializer/TopNCounterSerializer.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/measure/serializer/TopNCounterSerializer.java
@@ -18,44 +18,92 @@
package org.apache.kylin.metadata.measure.serializer;
-import org.apache.kylin.common.hll.HyperLogLogPlusCounter;
+import com.google.common.collect.Lists;
+import org.apache.kylin.common.topn.DoubleDeltaSerializer;
import org.apache.kylin.common.topn.TopNCounter;
import org.apache.kylin.common.util.ByteArray;
+import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.metadata.model.DataType;
-import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.List;
/**
*
*/
public class TopNCounterSerializer extends DataTypeSerializer<TopNCounter<ByteArray>> {
- // be thread-safe and avoid repeated obj creation
- private ThreadLocal<TopNCounter<ByteArray>> current = new ThreadLocal<TopNCounter<ByteArray>>();
+ private DoubleDeltaSerializer dds = new DoubleDeltaSerializer();
+
+ private int precision;
+
+ public TopNCounterSerializer(DataType dataType) {
+ this.precision = dataType.getPrecision();
+ }
@Override
public int peekLength(ByteBuffer in) {
- return 0;
+ int mark = in.position();
+ int capacity = in.getInt();
+ int size = in.getInt();
+ int keyLength = in.getInt();
+ dds.deserialize(in);
+ int len = in.position() - mark + keyLength * size;
+ in.position(mark);
+ return len;
}
@Override
public int maxLength() {
- return 0;
+ return precision * TopNCounter.EXTRA_SPACE_RATE * (4 + 8);
}
@Override
public TopNCounter<ByteArray> valueOf(byte[] value) {
- return null;
+ ByteBuffer buffer = ByteBuffer.wrap(value);
+ int sizeOfId = buffer.getInt();
+ int keyEncodedValue = buffer.getInt();
+ double counter = buffer.getDouble();
+
+ ByteArray key = new ByteArray(sizeOfId);
+ BytesUtil.writeUnsigned(keyEncodedValue, key.array(), 0, sizeOfId);
+
+ TopNCounter<ByteArray> topNCounter = new TopNCounter<ByteArray>(precision * TopNCounter.EXTRA_SPACE_RATE);
+ topNCounter.offer(key, counter);
+ return topNCounter;
}
@Override
public void serialize(TopNCounter<ByteArray> value, ByteBuffer out) {
+ double[] counters = value.getCounters();
+ List<ByteArray> items = value.getItems();
+ int keyLength = items.size() > 0 ? items.get(0).length() : 0;
+ out.putInt(value.getCapacity());
+ out.putInt(value.size());
+ out.putInt(keyLength);
+ dds.serialize(counters, out);
+ for (ByteArray item : items) {
+ out.put(item.array());
+ }
}
@Override
public TopNCounter<ByteArray> deserialize(ByteBuffer in) {
- return null;
+ int capacity = in.getInt();
+ int size = in.getInt();
+ int keyLength = in.getInt();
+ double[] counters = dds.deserialize(in);
+ List<ByteArray> items = Lists.newArrayList();
+
+ for(int i=0; i<size; i++) {
+ ByteArray byteArray = new ByteArray(keyLength);
+ in.get(byteArray.array());
+ items.add(byteArray);
+ }
+
+ TopNCounter<ByteArray> counter = new TopNCounter<ByteArray>(capacity);
+ counter.fromExternal(size, counters, items);
+ return counter;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/core-metadata/src/main/java/org/apache/kylin/metadata/model/ParameterDesc.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/ParameterDesc.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/ParameterDesc.java
index bdb4b91..9773b84 100644
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/model/ParameterDesc.java
+++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/ParameterDesc.java
@@ -40,8 +40,8 @@ public class ParameterDesc {
@JsonProperty("value")
private String value;
- @JsonProperty("counter")
- private String counter;
+ @JsonProperty("displaycolumn")
+ private String displayColumn;
private List<TblColRef> colRefs;
@@ -65,12 +65,12 @@ public class ParameterDesc {
this.value = value;
}
- public String getCounter() {
- return counter;
+ public String getDisplayColumn() {
+ return displayColumn;
}
- public void setCounter(String counter) {
- this.counter = counter;
+ public void setDisplayColumn(String displayColumn) {
+ this.displayColumn = displayColumn;
}
public List<TblColRef> getColRefs() {
@@ -102,7 +102,7 @@ public class ParameterDesc {
ParameterDesc that = (ParameterDesc) o;
- if (counter != null ? !counter.equals(that.counter) : that.counter != null) return false;
+ if (displayColumn != null ? !displayColumn.equals(that.displayColumn) : that.displayColumn != null) return false;
if (type != null ? !type.equals(that.type) : that.type != null) return false;
if (value != null ? !value.equals(that.value) : that.value != null) return false;
@@ -113,13 +113,13 @@ public class ParameterDesc {
public int hashCode() {
int result = type != null ? type.hashCode() : 0;
result = 31 * result + (value != null ? value.hashCode() : 0);
- result = 31 * result + (counter != null ? counter.hashCode() : 0);
+ result = 31 * result + (displayColumn != null ? displayColumn.hashCode() : 0);
return result;
}
@Override
public String toString() {
- return "ParameterDesc [type=" + type + ", value=" + value + ", counter=" + counter + "]";
+ return "ParameterDesc [type=" + type + ", value=" + value + ", displayColumn=" + displayColumn + "]";
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidMapperBase.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidMapperBase.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidMapperBase.java
index c2d0f92..ed1fd4a 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidMapperBase.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidMapperBase.java
@@ -4,7 +4,9 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
+import com.google.common.collect.Maps;
import org.apache.hadoop.io.Text;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.Bytes;
@@ -18,14 +20,12 @@ import org.apache.kylin.cube.kv.AbstractRowKeyEncoder;
import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.cube.model.CubeJoinedFlatTableDesc;
+import org.apache.kylin.dict.Dictionary;
import org.apache.kylin.engine.mr.KylinMapper;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
import org.apache.kylin.metadata.measure.MeasureCodec;
-import org.apache.kylin.metadata.model.FunctionDesc;
-import org.apache.kylin.metadata.model.MeasureDesc;
-import org.apache.kylin.metadata.model.ParameterDesc;
-import org.apache.kylin.metadata.model.SegmentStatusEnum;
+import org.apache.kylin.metadata.model.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -57,6 +57,7 @@ public class BaseCuboidMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VAL
private Text outputKey = new Text();
private Text outputValue = new Text();
private ByteBuffer valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
+ private Map<Integer, Dictionary<String>> topNDisplayColDictMap;
@Override
protected void setup(Context context) throws IOException {
@@ -91,8 +92,24 @@ public class BaseCuboidMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VAL
int colCount = cubeDesc.getRowkey().getRowKeyColumns().length;
keyBytesBuf = new byte[colCount][];
+ initTopNDisplayColDictionaryMap();
initNullBytes();
}
+
+ private void initTopNDisplayColDictionaryMap() {
+ topNDisplayColDictMap = Maps.newHashMap();
+ for (int measureIdx = 0; measureIdx < measures.length; measureIdx++) {
+ MeasureDesc measureDesc = cubeDesc.getMeasures().get(measureIdx);
+ FunctionDesc func = measureDesc.getFunction();
+ if (func.isTopN()) {
+ int[] flatTableIdx = intermediateTableDesc.getMeasureColumnIndexes()[measureIdx];
+ int displayColIdx = flatTableIdx[flatTableIdx.length - 1];
+ TblColRef displayCol = func.getParameter().getColRefs().get(flatTableIdx.length - 1);
+ Dictionary<String> dictionary = (Dictionary<String>)cubeSegment.getDictionary(displayCol);
+ topNDisplayColDictMap.put(displayColIdx, dictionary);
+ }
+ }
+ }
private void initNullBytes() {
nullBytes = Lists.newArrayList();
@@ -147,27 +164,49 @@ public class BaseCuboidMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VAL
// constant
if (flatTableIdx == null) {
result = Bytes.toBytes(paramDesc.getValue());
+ }
+ // count and count distinct
+ else if (func.isCount() || func.isHolisticCountDistinct()) {
+ // note for holistic count distinct, this value will be ignored
+ result = ONE;
}
- // column values
+ // topN, need encode the key column
+ else if(func.isTopN()) {
+ // encode the key column with dict, and get the counter column;
+ int keyColIndex = flatTableIdx[flatTableIdx.length - 1];
+ Dictionary<String> displayColDict = topNDisplayColDictMap.get(keyColIndex);
+ int keyColEncoded = displayColDict.getIdFromValue(Bytes.toString(splitBuffers[keyColIndex].value));
+ valueBuf.clear();
+ valueBuf.putInt(displayColDict.getSizeOfId());
+ valueBuf.putInt(keyColEncoded);
+ if (flatTableIdx.length == 1) {
+ // only displayCol, use 1.0 as counter
+ valueBuf.putDouble(1.0);
+ } else {
+ // get the counter column value
+ valueBuf.putDouble(Double.valueOf(Bytes.toString(splitBuffers[flatTableIdx[0]].value)));
+ }
+
+ result = valueBuf.array();
+
+ }
+ // normal case, concat column values
else {
- // for multiple columns, their values are joined
- for (int i = 0; i < flatTableIdx.length; i++) {
- SplittedBytes split = splitBuffers[flatTableIdx[i]];
- if (result == null) {
- result = Arrays.copyOf(split.value, split.length);
- } else {
- byte[] newResult = new byte[result.length + split.length];
- System.arraycopy(result, 0, newResult, 0, result.length);
- System.arraycopy(split.value, 0, newResult, result.length, split.length);
- result = newResult;
+ // for multiple columns, their values are joined
+ for (int i = 0; i < flatTableIdx.length; i++) {
+ SplittedBytes split = splitBuffers[flatTableIdx[i]];
+ if (result == null) {
+ result = Arrays.copyOf(split.value, split.length);
+ } else {
+ byte[] newResult = new byte[result.length + split.length];
+ System.arraycopy(result, 0, newResult, 0, result.length);
+ System.arraycopy(split.value, 0, newResult, result.length, split.length);
+ result = newResult;
+ }
}
}
- }
- if (func.isCount() || func.isHolisticCountDistinct()) {
- // note for holistic count distinct, this value will be ignored
- result = ONE;
- }
+
if (isNull(result)) {
result = null;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/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 098373a..b97c88a 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
@@ -1,10 +1,5 @@
package org.apache.kylin.engine.mr.steps;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
@@ -14,7 +9,7 @@ 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.RowKeyDesc;
+import org.apache.kylin.cube.model.CubeJoinedFlatTableDesc;
import org.apache.kylin.dict.DictionaryManager;
import org.apache.kylin.engine.mr.IMRInput.IMRTableInputFormat;
import org.apache.kylin.engine.mr.KylinMapper;
@@ -24,6 +19,11 @@ import org.apache.kylin.engine.mr.common.BatchConstants;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
import org.apache.kylin.metadata.model.TblColRef;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
/**
*/
public class FactDistinctColumnsMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VALUEIN, LongWritable, Text> {
@@ -33,7 +33,7 @@ public class FactDistinctColumnsMapperBase<KEYIN, VALUEIN> extends KylinMapper<K
protected CubeSegment cubeSeg;
protected CubeDesc cubeDesc;
protected long baseCuboidId;
- protected List<TblColRef> columns;
+ protected List<TblColRef> dictionaryColumns;
protected ArrayList<Integer> factDictCols;
protected IMRTableInputFormat flatTableInputFormat;
@@ -41,6 +41,9 @@ public class FactDistinctColumnsMapperBase<KEYIN, VALUEIN> extends KylinMapper<K
protected Text outputValue = new Text();
protected int errorRecordCounter = 0;
+ protected CubeJoinedFlatTableDesc intermediateTableDesc;
+ protected int[] dictionaryColumnIndex;
+
@Override
protected void setup(Context context) throws IOException {
Configuration conf = context.getConfiguration();
@@ -52,23 +55,30 @@ public class FactDistinctColumnsMapperBase<KEYIN, VALUEIN> extends KylinMapper<K
cubeSeg = cube.getSegment(conf.get(BatchConstants.CFG_CUBE_SEGMENT_NAME), SegmentStatusEnum.NEW);
cubeDesc = cube.getDescriptor();
baseCuboidId = Cuboid.getBaseCuboidId(cubeDesc);
- columns = Cuboid.findById(cubeDesc, baseCuboidId).getColumns();
+ dictionaryColumns = cubeDesc.getAllColumnsNeedDictionary();
factDictCols = new ArrayList<Integer>();
- RowKeyDesc rowKey = cubeDesc.getRowkey();
DictionaryManager dictMgr = DictionaryManager.getInstance(config);
- for (int i = 0; i < columns.size(); i++) {
- TblColRef col = columns.get(i);
- if (!rowKey.isUseDictionary(col))
- continue;
+ for (int i = 0; i < dictionaryColumns.size(); i++) {
+ TblColRef col = dictionaryColumns.get(i);
- String scanTable = dictMgr.decideSourceData(cubeDesc.getModel(), cubeDesc.getRowkey().getDictionary(col), col).getTable();
+ String scanTable = dictMgr.decideSourceData(cubeDesc.getModel(), "true", col).getTable();
if (cubeDesc.getModel().isFactTable(scanTable)) {
factDictCols.add(i);
}
}
flatTableInputFormat = MRUtil.getBatchCubingInputSide(cubeSeg).getFlatTableInputFormat();
+
+ intermediateTableDesc = new CubeJoinedFlatTableDesc(cubeDesc, null);
+ dictionaryColumnIndex = new int[factDictCols.size()];
+ for (int i = 0; i < factDictCols.size(); i++) {
+ Integer column = factDictCols.get(i);
+ TblColRef colRef = dictionaryColumns.get(column);
+ int columnIndexOnFlatTbl = intermediateTableDesc.getColumnIndex(colRef);
+ dictionaryColumnIndex[i] = columnIndexOnFlatTbl;
+ }
+
}
protected void handleErrorRecord(String[] record, Exception ex) throws IOException {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/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 fcc12e4..85312ff 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
@@ -77,9 +77,7 @@ public class FactDistinctColumnsReducer extends KylinReducer<LongWritable, Text,
CubeInstance cube = CubeManager.getInstance(config).getCube(cubeName);
cubeDesc = cube.getDescriptor();
- baseCuboidId = Cuboid.getBaseCuboidId(cubeDesc);
- Cuboid baseCuboid = Cuboid.findById(cubeDesc, baseCuboidId);
- columnList = baseCuboid.getColumns();
+ columnList = cubeDesc.getAllColumnsNeedDictionary();
collectStatistics = Boolean.parseBoolean(conf.get(BatchConstants.CFG_STATISTICS_ENABLED));
statisticsOutput = conf.get(BatchConstants.CFG_STATISTICS_OUTPUT);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/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 86b3605..e43d5d1 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
@@ -23,7 +23,9 @@ import java.nio.ByteBuffer;
import java.util.BitSet;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
+import com.google.common.collect.Maps;
import org.apache.kylin.common.hll.HyperLogLogPlusCounter;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.Bytes;
@@ -36,14 +38,13 @@ 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.model.TblColRef;
/**
* @author yangli9
*/
public class FactDistinctHiveColumnsMapper<KEYIN> extends FactDistinctColumnsMapperBase<KEYIN, Object> {
- private CubeJoinedFlatTableDesc intermediateTableDesc;
-
protected boolean collectStatistics = false;
protected CuboidScheduler cuboidScheduler = null;
protected int nRowKey;
@@ -58,7 +59,8 @@ public class FactDistinctHiveColumnsMapper<KEYIN> extends FactDistinctColumnsMap
@Override
protected void setup(Context context) throws IOException {
super.setup(context);
- intermediateTableDesc = new CubeJoinedFlatTableDesc(cubeDesc, null);
+
+
collectStatistics = Boolean.parseBoolean(context.getConfiguration().get(BatchConstants.CFG_STATISTICS_ENABLED));
if (collectStatistics) {
SAMPING_PERCENTAGE = Integer.parseInt(context.getConfiguration().get(BatchConstants.CFG_STATISTICS_SAMPLING_PERCENT, "5"));
@@ -111,9 +113,9 @@ public class FactDistinctHiveColumnsMapper<KEYIN> extends FactDistinctColumnsMap
public void map(KEYIN key, Object record, Context context) throws IOException, InterruptedException {
String[] row = flatTableInputFormat.parseMapperInput(record);
try {
- for (int i : factDictCols) {
- outputKey.set((long) i);
- String fieldValue = row[intermediateTableDesc.getRowKeyColumnIndexes()[i]];
+ for (int i = 0; i < factDictCols.size(); i++) {
+ outputKey.set((long) factDictCols.get(i));
+ String fieldValue = row[dictionaryColumnIndex[i]];
if (fieldValue == null)
continue;
byte[] bytes = Bytes.toBytes(fieldValue);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/examples/test_case_data/localmeta/cube/test_kylin_cube_topn.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube/test_kylin_cube_topn.json b/examples/test_case_data/localmeta/cube/test_kylin_cube_topn.json
new file mode 100644
index 0000000..903fc15
--- /dev/null
+++ b/examples/test_case_data/localmeta/cube/test_kylin_cube_topn.json
@@ -0,0 +1,10 @@
+{
+ "uuid" : "33354455-a33e-4b69-83dd-0bb8b1f8c53b",
+ "last_modified" : 0,
+ "name" : "test_kylin_cube_topn",
+ "owner" : null,
+ "version" : null,
+ "descriptor" : "test_kylin_cube_topn_desc",
+ "segments" : [ ],
+ "create_time" : null
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/db5952cf/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json
----------------------------------------------------------------------
diff --git a/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json
new file mode 100644
index 0000000..2e0e376
--- /dev/null
+++ b/examples/test_case_data/localmeta/cube_desc/test_kylin_cube_topn_desc.json
@@ -0,0 +1,146 @@
+{
+ "uuid": "4334a905-1fc6-4f67-985c-38fa5aeafd92",
+ "name": "test_kylin_cube_topn_desc",
+ "description": null,
+ "dimensions": [
+ {
+ "id": 0,
+ "name": "CAL_DT",
+ "table": "EDW.TEST_CAL_DT",
+ "column": null,
+ "derived": [
+ "WEEK_BEG_DT"
+ ],
+ "hierarchy": false
+ }
+ ],
+ "measures": [
+ {
+ "id": 1,
+ "name": "GMV_SUM",
+ "function": {
+ "expression": "SUM",
+ "parameter": {
+ "type": "column",
+ "value": "PRICE"
+ },
+ "returntype": "decimal(19,4)"
+ },
+ "dependent_measure_ref": null
+ },
+ {
+ "id": 2,
+ "name": "GMV_MIN",
+ "function": {
+ "expression": "MIN",
+ "parameter": {
+ "type": "column",
+ "value": "PRICE"
+ },
+ "returntype": "decimal(19,4)"
+ },
+ "dependent_measure_ref": null
+ },
+ {
+ "id": 3,
+ "name": "GMV_MAX",
+ "function": {
+ "expression": "MAX",
+ "parameter": {
+ "type": "column",
+ "value": "PRICE"
+ },
+ "returntype": "decimal(19,4)"
+ },
+ "dependent_measure_ref": null
+ },
+ {
+ "id": 4,
+ "name": "TRANS_CNT",
+ "function": {
+ "expression": "COUNT",
+ "parameter": {
+ "type": "constant",
+ "value": "1"
+ },
+ "returntype": "bigint"
+ },
+ "dependent_measure_ref": null
+ },
+ {
+ "id": 5,
+ "name": "ITEM_COUNT_SUM",
+ "function": {
+ "expression": "SUM",
+ "parameter": {
+ "type": "column",
+ "value": "ITEM_COUNT"
+ },
+ "returntype": "bigint"
+ },
+ "dependent_measure_ref": null
+ },
+ {
+ "id": 6,
+ "name": "TOP_SELLER",
+ "function": {
+ "expression": "TOP_N",
+ "parameter": {
+ "type": "column",
+ "value": "PRICE",
+ "displaycolumn": "seller_id"
+ },
+ "returntype": "topn(100)"
+ },
+ "dependent_measure_ref": null
+ }
+ ],
+ "rowkey": {
+ "rowkey_columns": [
+ {
+ "column": "cal_dt",
+ "length": 0,
+ "dictionary": "true",
+ "mandatory": false
+ }
+ ],
+ "aggregation_groups": [
+ [
+ "cal_dt"
+ ]
+ ]
+ },
+ "last_modified": 1422435345330,
+ "model_name": "test_kylin_inner_join_model_desc",
+ "null_string": null,
+ "hbase_mapping": {
+ "column_family": [
+ {
+ "name": "f1",
+ "columns": [
+ {
+ "qualifier": "m",
+ "measure_refs": [
+ "gmv_sum",
+ "gmv_min",
+ "gmv_max",
+ "trans_cnt",
+ "item_count_sum"
+ ]
+ }
+ ]
+ }, {
+ "name": "f2",
+ "columns": [
+ {
+ "qualifier": "m",
+ "measure_refs": [
+ "top_seller"
+ ]
+ }
+ ]
+ }
+ ]
+ },
+ "notify_list": null
+}
\ No newline at end of file