This is an automated email from the ASF dual-hosted git repository.
xxyu pushed a commit to branch kylin-on-parquet-v2
in repository https://gitbox.apache.org/repos/asf/kylin.git
The following commit(s) were added to refs/heads/kylin-on-parquet-v2 by this
push:
new 914b97f KYLIN-5011 Detect and scatter skewed data in dict encoding
step (#1662)
914b97f is described below
commit 914b97f5cf2347030525140038d060178b93f955
Author: zhengshengjun <74281684+zhengsheng...@users.noreply.github.com>
AuthorDate: Sun Jun 27 21:34:55 2021 +0800
KYLIN-5011 Detect and scatter skewed data in dict encoding step (#1662)
Co-authored-by: Xiaoxiang Yu <x...@apache.org>
---
.../org/apache/kylin/common/KylinConfigBase.java | 28 ++++++++++
examples/test_case_data/localmeta/kylin.properties | 3 +-
.../org/apache/spark/dict/NBucketDictionary.java | 21 +++++++-
.../org/apache/spark/dict/NGlobalDictionary.java | 6 ++-
.../org/apache/spark/sql/KylinFunctions.scala | 6 ++-
.../catalyst/expressions/KylinExpresssions.scala | 60 ++++++++++++++++++++-
.../engine/spark/builder/CreateFlatTable.scala | 5 +-
.../engine/spark/builder/CubeTableEncoder.scala | 61 +++++++++++++++++++---
.../engine/spark/job/ParentSourceChooser.scala | 2 +-
.../engine/spark/LocalWithSparkSessionTest.java | 2 +-
.../engine/spark/builder/TestCreateFlatTable.scala | 10 +++-
11 files changed, 186 insertions(+), 18 deletions(-)
diff --git
a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
index ce5d5e8..39f8eae 100644
--- a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
@@ -3168,4 +3168,32 @@ public abstract class KylinConfigBase implements
Serializable {
public boolean rePartitionEncodedDatasetWithRowKey() {
return
Boolean.valueOf(getOptional("kylin.engine.spark.repartition.encoded.dataset",
"false"));
}
+
+ /*
+ * Detect dataset skew in dictionary encode step.
+ * */
+ public boolean detectDataSkewInDictEncodingEnabled() {
+ return
Boolean.valueOf(getOptional("kylin.dictionary.detect.data.skew.in.encoding",
"false"));
+ }
+
+ /*
+ * In some data skew cases, the repartition step during dictionary encoding
will be slow.
+ * We can choose to sample from the dataset to detect skewed. This
configuration is used to set the sample rate.
+ * */
+ public double sampleRateInEncodingSkewDetection() {
+ return
Double.valueOf(getOptional("kylin.dictionary.detect.data.skew.sample.rate",
"0.1"));
+ }
+
+ /*
+ * In KYLIN4, dictionaries are hashed into several buckets, column data are
repartitioned by the same hash algorithm
+ * during encoding step too. In data skew cases, the repartition step will
be very slow. Kylin will automatically
+ * sample from the source to detect skewed data and repartition these
skewed data to random partitions.
+ * This configuration is used to set the skew data threshhold, valued from
0 to 1.
+ * e.g.
+ * if you set this value to 0.05, for each value that takes up more than
5% percent of the total will be regarded
+ * as skew data, as a result the skewed data will be no more than 20
records
+ * */
+ public double skewPercentageThreshHold() {
+ return
Double.valueOf(getOptional("kylin.dictionary.data.skew.percentage.threshhold",
"0.05"));
+ }
}
diff --git a/examples/test_case_data/localmeta/kylin.properties
b/examples/test_case_data/localmeta/kylin.properties
index d17ce8c..4b015c4 100644
--- a/examples/test_case_data/localmeta/kylin.properties
+++ b/examples/test_case_data/localmeta/kylin.properties
@@ -162,4 +162,5 @@ kylin.source.jdbc.pass=
kylin.query.auto-sparder-context=false
kylin.metrics.query-cache.expire-seconds=5
-kylin.metrics.query-cache.max-entries=2
\ No newline at end of file
+kylin.metrics.query-cache.max-entries=2
+kylin.dictionary.detect.data.skew.in.encoding=true
\ No newline at end of file
diff --git
a/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/dict/NBucketDictionary.java
b/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/dict/NBucketDictionary.java
index bf2a351..2e310ca 100644
---
a/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/dict/NBucketDictionary.java
+++
b/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/dict/NBucketDictionary.java
@@ -19,6 +19,12 @@ package org.apache.spark.dict;
import java.io.IOException;
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,8 +42,9 @@ public class NBucketDictionary {
private Object2LongMap<String> absoluteDictMap;
// Relative dictionary needs to calculate dictionary code according to
NGlobalDictMetaInfo's bucketOffsets
private Object2LongMap<String> relativeDictMap;
+ private Object2LongMap<String> skewedDictMap;
- NBucketDictionary(String baseDir, String workingDir, int bucketId,
NGlobalDictMetaInfo metainfo)
+ NBucketDictionary(String baseDir, String workingDir, int bucketId,
NGlobalDictMetaInfo metainfo, String skewDictStorageFile)
throws IOException {
this.workingDir = workingDir;
this.bucketId = bucketId;
@@ -49,6 +56,15 @@ public class NBucketDictionary {
this.absoluteDictMap =
globalDictStore.getBucketDict(versions[versions.length - 1], metainfo,
bucketId);
}
this.relativeDictMap = new Object2LongOpenHashMap<>();
+ if (!StringUtils.isEmpty(skewDictStorageFile)) {
+ FileSystem fs = FileSystem.get(new Configuration());
+ if (fs.exists(new Path(skewDictStorageFile))) {
+ Kryo kryo = new Kryo();
+ Input input = new Input(fs.open(new
Path(skewDictStorageFile)));
+ skewedDictMap = (Object2LongMap<String>)
kryo.readClassAndObject(input);
+ input.close();
+ }
+ }
}
NBucketDictionary(String workingDir) {
@@ -72,6 +88,9 @@ public class NBucketDictionary {
}
public long encode(Object value) {
+ if (null != skewedDictMap &&
skewedDictMap.containsKey(value.toString())) {
+ return skewedDictMap.getLong(value.toString());
+ }
return absoluteDictMap.getLong(value.toString());
}
diff --git
a/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/dict/NGlobalDictionary.java
b/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/dict/NGlobalDictionary.java
index 651387d..a74022c 100644
---
a/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/dict/NGlobalDictionary.java
+++
b/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/dict/NGlobalDictionary.java
@@ -39,6 +39,7 @@ public class NGlobalDictionary implements Serializable {
private String sourceTable;
private String sourceColumn;
private boolean isFirst = true;
+ private String skewDictStorageFile;
public String getResourceDir() {
return "/" + project + HadoopUtil.GLOBAL_DICT_STORAGE_ROOT + "/" +
sourceTable + "/" + sourceColumn + "/";
@@ -71,13 +72,16 @@ public class NGlobalDictionary implements Serializable {
if (metadata != null) {
isFirst = false;
}
+ if (dictInfo.length >= 5) {
+ skewDictStorageFile = dictInfo[4];
+ }
}
public NBucketDictionary loadBucketDictionary(int bucketId) throws
IOException {
if (null == metadata) {
metadata = getMetaInfo();
}
- return new NBucketDictionary(baseDir, getWorkingDir(), bucketId,
metadata);
+ return new NBucketDictionary(baseDir, getWorkingDir(), bucketId,
metadata, skewDictStorageFile);
}
public NBucketDictionary createNewBucketDictionary() {
diff --git
a/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/sql/KylinFunctions.scala
b/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/sql/KylinFunctions.scala
index 8e112c1..63c90dc 100644
---
a/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/sql/KylinFunctions.scala
+++
b/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/sql/KylinFunctions.scala
@@ -24,7 +24,7 @@ import
org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCo
import org.apache.spark.sql.types._
import org.apache.spark.sql.catalyst.expressions.{ApproxCountDistinctDecode,
BinaryExpression,
DictEncode, Expression, ExpressionInfo, ExpressionUtils,
ImplicitCastInputTypes, In,
- KylinAddMonths, Like, Literal, PreciseCountDistinctDecode, RoundBase,
SplitPart, Sum0,
+ KylinAddMonths, Like, Literal, PreciseCountDistinctDecode, RoundBase,
ScatterSkewData, SplitPart, Sum0,
TimestampAdd, TimestampDiff, Truncate, UnaryExpression}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction
import org.apache.spark.sql.udaf.{ApproxCountDistinct, IntersectCount,
PreciseCountDistinct}
@@ -44,6 +44,10 @@ object KylinFunctions {
Column(DictEncode(column.expr, dictParams.expr, bucketSize.expr))
}
+ def scatter_skew_data(column: Column, skewDataStorage: Column): Column = {
+ Column(ScatterSkewData(column.expr, skewDataStorage.expr))
+ }
+
// special lit for KYLIN.
def k_lit(literal: Any): Column = literal match {
case c: Column => c
diff --git
a/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/sql/catalyst/expressions/KylinExpresssions.scala
b/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/sql/catalyst/expressions/KylinExpresssions.scala
index a1a45fa..d25896c 100644
---
a/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/sql/catalyst/expressions/KylinExpresssions.scala
+++
b/kylin-spark-project/kylin-spark-common/src/main/scala/org/apache/spark/sql/catalyst/expressions/KylinExpresssions.scala
@@ -24,7 +24,7 @@ import org.apache.spark.dict.{NBucketDictionary,
NGlobalDictionary}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
import org.apache.spark.sql.catalyst.expressions.codegen.Block._
-import org.apache.spark.sql.catalyst.expressions.codegen.{CodeGenerator,
CodegenContext, ExprCode}
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodeGenerator,
CodegenContext, ExprCode, FalseLiteral}
import org.apache.spark.sql.types._
import org.roaringbitmap.longlong.Roaring64NavigableMap
@@ -484,4 +484,62 @@ case class ApproxCountDistinctDecode(_left: Expression,
_right: Expression)
override def dataType: DataType = LongType
override def prettyName: String = "approx_count_distinct_decode"
+}
+
+case class ScatterSkewData(left: Expression, right: Expression) extends
BinaryExpression with ExpectsInputTypes {
+ override protected def doGenCode(ctx: CodegenContext, ev: ExprCode) :
ExprCode = {
+
+ val rand = ctx.addMutableState("java.util.Random", "rand")
+ val skewData =
ctx.addMutableState("it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap",
+ "skewData")
+ val skewDataStorage = right.simpleString
+
+ val initParamsFuncName = ctx.addNewFunction(s"initParams",
+ s"""
+ | private void initParams() {
+ | ${rand} = new java.util.Random();
+ | com.esotericsoftware.kryo.Kryo kryo = new
com.esotericsoftware.kryo.Kryo();
+ | try {
+ | org.apache.hadoop.fs.FileSystem fs =
org.apache.hadoop.fs.FileSystem.get(new org.apache.hadoop.conf.Configuration());
+ | if (fs.exists(new
org.apache.hadoop.fs.Path("${skewDataStorage}"))) {
+ | com.esotericsoftware.kryo.io.Input input = new
com.esotericsoftware.kryo.io.Input(
+ | fs.open(new
org.apache.hadoop.fs.Path("${skewDataStorage}")));
+ | ${skewData} =
(it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap<String>)
kryo.readClassAndObject(input);
+ | input.close();
+ | }
+ | } catch (java.io.IOException e) {
+ | throw new java.lang.RuntimeException(e);
+ | }
+ | }
+ """.stripMargin)
+
+ val addSalt = ctx.addNewFunction(s"addSalt",
+ s"""
+ | private org.apache.spark.unsafe.types.UTF8String
addSalt(org.apache.spark.unsafe.types.UTF8String val) {
+ | if (null != ${skewData} && (null == val ||
${skewData}.containsKey(val.toString()))) {
+ | return org.apache.spark.unsafe.types.UTF8String.fromString(
+ | java.lang.Integer.toString(${rand}.nextInt()));
+ | } else {
+ | return val;
+ | }
+ | }
+ """.stripMargin)
+
+ ctx.addPartitionInitializationStatement(s"$initParamsFuncName();");
+
+ val leftGen = left.genCode(ctx)
+ val rightGen = right.genCode(ctx)
+ val resultCode = s"""${ev.value} = $addSalt(${leftGen.value});"""
+
+ ev.copy(code = code"""
+ ${leftGen.code}
+ ${rightGen.code}
+ ${CodeGenerator.javaType(dataType)} ${ev.value} =
${CodeGenerator.defaultValue(dataType)};
+ $resultCode""", isNull = FalseLiteral)
+
+ }
+
+ override def dataType: DataType = StringType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType,
AnyDataType)
}
\ No newline at end of file
diff --git
a/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/builder/CreateFlatTable.scala
b/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/builder/CreateFlatTable.scala
index 6e8dc22..3bfd519 100644
---
a/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/builder/CreateFlatTable.scala
+++
b/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/builder/CreateFlatTable.scala
@@ -36,7 +36,8 @@ import scala.collection.JavaConverters._
class CreateFlatTable(val seg: SegmentInfo,
val toBuildTree: SpanningTree,
val ss: SparkSession,
- val sourceInfo: NBuildSourceInfo) extends Logging {
+ val sourceInfo: NBuildSourceInfo,
+ val jobId: String) extends Logging {
import org.apache.kylin.engine.spark.builder.CreateFlatTable._
@@ -108,7 +109,7 @@ class CreateFlatTable(val seg: SegmentInfo,
val matchedCols = filterCols(ds, encodeCols)
var encodeDs = ds
if (!matchedCols.isEmpty) {
- encodeDs = CubeTableEncoder.encodeTable(ds, seg, matchedCols.asJava)
+ encodeDs = CubeTableEncoder.encodeTable(ds, seg, matchedCols.asJava,
jobId)
}
encodeDs
}
diff --git
a/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/builder/CubeTableEncoder.scala
b/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/builder/CubeTableEncoder.scala
index d2a34a8..8a5fe20 100644
---
a/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/builder/CubeTableEncoder.scala
+++
b/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/builder/CubeTableEncoder.scala
@@ -17,8 +17,13 @@
*/
package org.apache.kylin.engine.spark.builder
-import java.util
+import com.esotericsoftware.kryo.Kryo
+import com.esotericsoftware.kryo.io.Output
+import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{Path}
+import java.util
import org.apache.kylin.engine.spark.builder.CubeBuilderHelper.ENCODE_SUFFIX
import org.apache.kylin.engine.spark.job.NSparkCubingUtil._
import org.apache.kylin.engine.spark.metadata.{ColumnDesc, SegmentInfo}
@@ -27,7 +32,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.KylinFunctions._
import org.apache.spark.sql.functions.{col, _}
import org.apache.spark.sql.types.StringType
-import org.apache.spark.sql.{Dataset, Row}
+import org.apache.spark.sql.{Dataset, Row, functions}
import org.apache.spark.utils.SparkVersionUtils
import scala.collection.JavaConverters._
@@ -35,7 +40,7 @@ import scala.collection.mutable._
object CubeTableEncoder extends Logging {
- def encodeTable(ds: Dataset[Row], seg: SegmentInfo, cols:
util.Set[ColumnDesc]): Dataset[Row] = {
+ def encodeTable(ds: Dataset[Row], seg: SegmentInfo, cols:
util.Set[ColumnDesc], jobId: String): Dataset[Row] = {
if (SparkVersionUtils.isLessThanSparkVersion("2.4", true)) {
assert(!ds.sparkSession.conf.get("spark.sql.adaptive.enabled",
"false").toBoolean,
"Parameter 'spark.sql.adaptive.enabled' must be false when encode
tables.")
@@ -61,15 +66,57 @@ object CubeTableEncoder extends Logging {
val encodeColRef = convertFromDot(ref.identity)
val columnIndex = structType.fieldIndex(encodeColRef)
- val dictParams = Array(seg.project, ref.tableAliasName,
ref.columnName, seg.kylinconf.getHdfsWorkingDirectory)
+ var dictParams = Array(seg.project, ref.tableAliasName,
ref.columnName, seg.kylinconf.getHdfsWorkingDirectory)
.mkString(SEPARATOR)
val aliasName =
structType.apply(columnIndex).name.concat(ENCODE_SUFFIX)
- val encodeCol = dict_encode(col(encodeColRef).cast(StringType),
lit(dictParams), lit(bucketSize).cast(StringType)).as(aliasName)
- val columns = partitionedDs.schema.map(ty => col(ty.name)) ++
Seq(encodeCol)
+ var encodeCol = dict_encode(col(encodeColRef).cast(StringType),
lit(dictParams), lit(bucketSize).cast(StringType)).as(aliasName)
+ val columns = partitionedDs.schema.map(ty => col(ty.name))
+
+ if (seg.kylinconf.detectDataSkewInDictEncodingEnabled()) {
+ //find skewed data in dict-encoding step
+ val castEncodeColRef = col(encodeColRef).cast(StringType)
+ val sampleData =
ds.select(castEncodeColRef).sample(seg.kylinconf.sampleRateInEncodingSkewDetection()).cache()
+ val totalCount = sampleData.count();
+ val skewDictStorage = new
Path(seg.kylinconf.getJobTmpDir(seg.project) +
+ "/" + jobId + "/skewed_data/" + ref.identity)
+ val skewedDict = new Object2LongOpenHashMap[String]()
+ sampleData.groupBy(encodeColRef)
+ .agg(functions.count(lit(1)).alias("count_value"))
+ .filter(col("count_value") > totalCount *
seg.kylinconf.skewPercentageThreshHold())
+ .repartition(enlargedBucketSize, castEncodeColRef)
+ .select(Seq(castEncodeColRef, encodeCol): _*)
+ .collect().foreach(row => skewedDict.put(row.getString(0),
row.getLong(1)));
+ sampleData.unpersist()
+
+ //save skewed data dict
+ if (skewedDict.size() > 0) {
+ val kryo = new Kryo()
+ val fs = skewDictStorage.getFileSystem(new Configuration())
+ if (fs.exists(skewDictStorage)) {
+ fs.delete(skewDictStorage, true)
+ }
+ val output = new Output(fs.create(skewDictStorage))
+ kryo.writeClassAndObject(output, skewedDict)
+ output.close()
+ //define repartition expression: repartition skewed data to random
partition
+ val scatterColumn = scatter_skew_data(castEncodeColRef,
lit(skewDictStorage.toString))
+ .alias("scatter_skew_data_" + ref.columnName)
+
+ //encode cuboid table with skewed data dictionary
+ dictParams = Array(seg.project, ref.tableAliasName,
ref.columnName, seg.kylinconf.getHdfsWorkingDirectory, skewDictStorage.toString)
+ .mkString(SEPARATOR)
+ encodeCol = dict_encode(col(encodeColRef).cast(StringType),
lit(dictParams), lit(bucketSize).cast(StringType)).alias(aliasName)
+
+ partitionedDs = partitionedDs.select(columns ++
Seq(scatterColumn): _*)
+ .repartition(enlargedBucketSize, col("scatter_skew_data_" +
ref.columnName))
+ .select(columns ++ Seq(encodeCol): _*)
+ return partitionedDs;
+ }
+ }
partitionedDs = partitionedDs
.repartition(enlargedBucketSize, col(encodeColRef).cast(StringType))
- .select(columns: _*)
+ .select(columns ++ Seq(encodeCol): _*)
}
)
diff --git
a/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/job/ParentSourceChooser.scala
b/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/job/ParentSourceChooser.scala
index bd54d0b..fdb04ab 100644
---
a/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/job/ParentSourceChooser.scala
+++
b/kylin-spark-project/kylin-spark-engine/src/main/scala/org/apache/kylin/engine/spark/job/ParentSourceChooser.scala
@@ -185,7 +185,7 @@ class ParentSourceChooser(
// sourceInfo.setViewFactTablePath(viewPath)
// val needJoin =
ParentSourceChooser.needJoinLookupTables(segInfo.getModel, toBuildTree)
- val flatTable = new CreateFlatTable(segInfo, toBuildTree, ss, sourceInfo)
+ val flatTable = new CreateFlatTable(segInfo, toBuildTree, ss, sourceInfo,
jobId)
val afterJoin: Dataset[Row] = flatTable.generateDataset(needEncoding, true)
sourceInfo.setFlatTableDS(afterJoin)
diff --git
a/kylin-spark-project/kylin-spark-engine/src/test/java/org/apache/kylin/engine/spark/LocalWithSparkSessionTest.java
b/kylin-spark-project/kylin-spark-engine/src/test/java/org/apache/kylin/engine/spark/LocalWithSparkSessionTest.java
index cfd84aa..ae2bb15 100644
---
a/kylin-spark-project/kylin-spark-engine/src/test/java/org/apache/kylin/engine/spark/LocalWithSparkSessionTest.java
+++
b/kylin-spark-project/kylin-spark-engine/src/test/java/org/apache/kylin/engine/spark/LocalWithSparkSessionTest.java
@@ -312,7 +312,7 @@ public class LocalWithSparkSessionTest extends
LocalFileMetadataTestCase impleme
CreateFlatTable flatTable = new CreateFlatTable(
MetadataConverter.getSegmentInfo(segment.getCubeInstance(),
segment.getUuid(),
- segment.getName(),
segment.getStorageLocationIdentifier()), null, ss, null);
+ segment.getName(),
segment.getStorageLocationIdentifier()), null, ss, null,
ss.sparkContext().applicationId());
Dataset<Row> ds = flatTable.generateDataset(false, true);
return ds;
}
diff --git
a/kylin-spark-project/kylin-spark-engine/src/test/scala/org/apache/kylin/engine/spark/builder/TestCreateFlatTable.scala
b/kylin-spark-project/kylin-spark-engine/src/test/scala/org/apache/kylin/engine/spark/builder/TestCreateFlatTable.scala
index f8dd610..df0273c 100644
---
a/kylin-spark-project/kylin-spark-engine/src/test/scala/org/apache/kylin/engine/spark/builder/TestCreateFlatTable.scala
+++
b/kylin-spark-project/kylin-spark-engine/src/test/scala/org/apache/kylin/engine/spark/builder/TestCreateFlatTable.scala
@@ -96,9 +96,14 @@ class TestCreateFlatTable extends SparderBaseFunSuite with
SharedSparkSession wi
val afterJoin1 = generateFlatTable(seg1, cube, true)
afterJoin1.collect()
+
if (SPARK_VERSION.startsWith("2.4")) {
val jobs = helper.getJobsByGroupId(groupId)
- Assert.assertEquals(jobs.length, 15)
+ if (seg1.getConfig.detectDataSkewInDictEncodingEnabled()) {
+ Assert.assertEquals(jobs.length, 18)
+ } else {
+ Assert.assertEquals(jobs.length, 15)
+ }
} else if (SPARK_VERSION.startsWith("3.1")) {
// in Spark 3.x, BroadcastExchangeExec overwrites job group ID
val jobs = helper.getJobsByGroupId(null)
@@ -140,7 +145,8 @@ class TestCreateFlatTable extends SparderBaseFunSuite with
SharedSparkSession wi
private def generateFlatTable(segment: CubeSegment, cube: CubeInstance,
needEncode: Boolean): Dataset[Row] = {
val seg = MetadataConverter.getSegmentInfo(segment.getCubeInstance,
segment.getUuid, segment.getName, segment.getStorageLocationIdentifier)
val spanningTree = new
ForestSpanningTree(JavaConversions.asJavaCollection(seg.toBuildLayouts))
- val flatTable = new CreateFlatTable(seg, spanningTree, spark, null)
+ //for test case there is no build job id
+ val flatTable = new CreateFlatTable(seg, spanningTree, spark, null,
spark.sparkContext.applicationId)
val afterJoin = flatTable.generateDataset(needEncode)
afterJoin
}
