| /* |
| * 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.griffin.measure.step.builder.dsl.transform |
| |
| import org.apache.griffin.measure.configuration.dqdefinition.RuleParam |
| import org.apache.griffin.measure.configuration.enums.FlattenType.{ |
| ArrayFlattenType, |
| EntriesFlattenType |
| } |
| import org.apache.griffin.measure.configuration.enums.OutputType._ |
| import org.apache.griffin.measure.configuration.enums.ProcessType._ |
| import org.apache.griffin.measure.context.DQContext |
| import org.apache.griffin.measure.step.DQStep |
| import org.apache.griffin.measure.step.builder.ConstantColumns |
| import org.apache.griffin.measure.step.builder.dsl.expr.{DistinctnessClause, _} |
| import org.apache.griffin.measure.step.builder.dsl.transform.analyzer.DistinctnessAnalyzer |
| import org.apache.griffin.measure.step.transform.SparkSqlTransformStep |
| import org.apache.griffin.measure.step.write.{ |
| DataSourceUpdateWriteStep, |
| MetricWriteStep, |
| RecordWriteStep |
| } |
| import org.apache.griffin.measure.utils.ParamUtil._ |
| |
| /** |
| * generate distinctness dq steps |
| */ |
| case class DistinctnessExpr2DQSteps(context: DQContext, expr: Expr, ruleParam: RuleParam) |
| extends Expr2DQSteps { |
| |
| private object DistinctnessKeys { |
| val _source = "source" |
| val _target = "target" |
| val _distinct = "distinct" |
| val _total = "total" |
| val _dup = "dup" |
| val _accu_dup = "accu_dup" |
| val _num = "num" |
| |
| val _duplicationArray = "duplication.array" |
| val _withAccumulate = "with.accumulate" |
| |
| val _recordEnable = "record.enable" |
| } |
| import DistinctnessKeys._ |
| |
| def getDQSteps: Seq[DQStep] = { |
| val details = ruleParam.getDetails |
| val distinctnessExpr = expr.asInstanceOf[DistinctnessClause] |
| |
| val sourceName = details.getString(_source, context.getDataSourceName(0)) |
| val targetName = details.getString(_target, context.getDataSourceName(1)) |
| val analyzer = DistinctnessAnalyzer(distinctnessExpr, sourceName) |
| |
| val procType = context.procType |
| val timestamp = context.contextId.timestamp |
| val dsTimeRanges = context.dataSourceTimeRanges |
| |
| val beginTmst = dsTimeRanges.get(sourceName).map(_.begin) match { |
| case Some(t) => t |
| case _ => throw new Exception(s"empty begin tmst from $sourceName") |
| } |
| val endTmst = dsTimeRanges.get(sourceName).map(_.end) match { |
| case Some(t) => t |
| case _ => throw new Exception(s"empty end tmst from $sourceName") |
| } |
| |
| val writeTimestampOpt = Some(endTmst) |
| |
| if (!context.runTimeTableRegister.existsTable(sourceName)) { |
| warn(s"[$timestamp] data source $sourceName not exists") |
| Nil |
| } else { |
| val withOlderTable = { |
| details.getBoolean(_withAccumulate, defValue = true) && |
| context.runTimeTableRegister.existsTable(targetName) |
| } |
| |
| val selClause = analyzer.selectionPairs |
| .map { pair => |
| val (expr, alias, _) = pair |
| s"${expr.desc} AS `$alias`" |
| } |
| .mkString(", ") |
| val distAliases = analyzer.selectionPairs.filter(_._3).map(_._2) |
| val distAliasesClause = distAliases.map(a => s"`$a`").mkString(", ") |
| val allAliases = analyzer.selectionPairs.map(_._2) |
| val allAliasesClause = allAliases.map(a => s"`$a`").mkString(", ") |
| val groupAliases = analyzer.selectionPairs.filter(!_._3).map(_._2) |
| val groupAliasesClause = groupAliases.map(a => s"`$a`").mkString(", ") |
| |
| // 1. source alias |
| val sourceAliasTableName = "__sourceAlias" |
| val sourceAliasSql = { |
| s"SELECT $selClause FROM `$sourceName`" |
| } |
| val sourceAliasTransStep = |
| SparkSqlTransformStep(sourceAliasTableName, sourceAliasSql, emptyMap, None, cache = true) |
| |
| // 2. total metric |
| val totalTableName = "__totalMetric" |
| val totalColName = details.getStringOrKey(_total) |
| val totalSql = { |
| s"SELECT COUNT(*) AS `$totalColName` FROM `$sourceAliasTableName`" |
| } |
| val totalMetricWriteStep = { |
| MetricWriteStep(totalColName, totalTableName, EntriesFlattenType, writeTimestampOpt) |
| } |
| val totalTransStep = |
| SparkSqlTransformStep(totalTableName, totalSql, emptyMap, Some(totalMetricWriteStep)) |
| totalTransStep.parentSteps += sourceAliasTransStep |
| |
| // 3. group by self |
| val selfGroupTableName = "__selfGroup" |
| val dupColName = details.getStringOrKey(_dup) |
| val accuDupColName = details.getStringOrKey(_accu_dup) |
| val selfGroupSql = { |
| s""" |
| |SELECT $distAliasesClause, (COUNT(*) - 1) AS `$dupColName`, |
| |TRUE AS `${ConstantColumns.distinct}` |
| |FROM `$sourceAliasTableName` GROUP BY $distAliasesClause |
| """.stripMargin |
| } |
| val selfGroupTransStep = |
| SparkSqlTransformStep(selfGroupTableName, selfGroupSql, emptyMap, None, cache = true) |
| selfGroupTransStep.parentSteps += sourceAliasTransStep |
| |
| val transSteps1 = totalTransStep :: selfGroupTransStep :: Nil |
| |
| val (transSteps2, dupCountTableName) = procType match { |
| case StreamingProcessType if withOlderTable => |
| // 4.0 update old data |
| val targetDsUpdateWriteStep = DataSourceUpdateWriteStep(targetName, targetName) |
| |
| // 4. older alias |
| val olderAliasTableName = "__older" |
| val olderAliasSql = { |
| s"SELECT $selClause FROM `$targetName` WHERE `${ConstantColumns.tmst}` <= $beginTmst" |
| } |
| val olderAliasTransStep = |
| SparkSqlTransformStep(olderAliasTableName, olderAliasSql, emptyMap) |
| |
| // 5. join with older data |
| val joinedTableName = "__joined" |
| val selfSelClause = (distAliases :+ dupColName) |
| .map { alias => |
| s"`$selfGroupTableName`.`$alias`" |
| } |
| .mkString(", ") |
| val onClause = distAliases |
| .map { alias => |
| s"coalesce(`$selfGroupTableName`.`$alias`, '') = coalesce(`$olderAliasTableName`.`$alias`, '')" |
| } |
| .mkString(" AND ") |
| val olderIsNull = distAliases |
| .map { alias => |
| s"`$olderAliasTableName`.`$alias` IS NULL" |
| } |
| .mkString(" AND ") |
| val joinedSql = { |
| s""" |
| |SELECT $selfSelClause, ($olderIsNull) AS `${ConstantColumns.distinct}` |
| |FROM `$olderAliasTableName` RIGHT JOIN `$selfGroupTableName` |
| |ON $onClause |
| """.stripMargin |
| } |
| val joinedTransStep = SparkSqlTransformStep(joinedTableName, joinedSql, emptyMap) |
| joinedTransStep.parentSteps += selfGroupTransStep |
| joinedTransStep.parentSteps += olderAliasTransStep |
| |
| // 6. group by joined data |
| val groupTableName = "__group" |
| val moreDupColName = "_more_dup" |
| val groupSql = { |
| s""" |
| |SELECT $distAliasesClause, `$dupColName`, `${ConstantColumns.distinct}`, |
| |COUNT(*) AS `$moreDupColName` |
| |FROM `$joinedTableName` |
| |GROUP BY $distAliasesClause, `$dupColName`, `${ConstantColumns.distinct}` |
| """.stripMargin |
| } |
| val groupTransStep = SparkSqlTransformStep(groupTableName, groupSql, emptyMap) |
| groupTransStep.parentSteps += joinedTransStep |
| |
| // 7. final duplicate count |
| val finalDupCountTableName = "__finalDupCount" |
| |
| /** |
| * dupColName: the duplicate count of duplicated items only occurs in new data, |
| * which means the distinct one in new data is also duplicate |
| * accuDupColName: the count of duplicated items accumulated in new data and old data, |
| * which means the accumulated distinct count in all data |
| * e.g.: new data [A, A, B, B, C, D], old data [A, A, B, C] |
| * selfGroupTable will be (A, 1, F), (B, 1, F), (C, 0, T), (D, 0, T) |
| * joinedTable will be (A, 1, F), (A, 1, F), (B, 1, F), (C, 0, F), (D, 0, T) |
| * groupTable will be (A, 1, F, 2), (B, 1, F, 1), (C, 0, F, 1), (D, 0, T, 1) |
| * finalDupCountTable will be (A, F, 2, 3), (B, F, 2, 2), (C, F, 1, 1), (D, T, 0, 0) |
| * The distinct result of new data only should be: (A, 2), (B, 2), (C, 1), (D, 0), |
| * which means in new data [A, A, B, B, C, D], [A, A, B, B, C] are all duplicated, |
| * only [D] is distinct |
| */ |
| val finalDupCountSql = { |
| s""" |
| |SELECT $distAliasesClause, `${ConstantColumns.distinct}`, |
| |CASE WHEN `${ConstantColumns.distinct}` THEN `$dupColName` |
| |ELSE (`$dupColName` + 1) END AS `$dupColName`, |
| |CASE WHEN `${ConstantColumns.distinct}` THEN `$dupColName` |
| |ELSE (`$dupColName` + `$moreDupColName`) END AS `$accuDupColName` |
| |FROM `$groupTableName` |
| """.stripMargin |
| } |
| val finalDupCountTransStep = |
| SparkSqlTransformStep( |
| finalDupCountTableName, |
| finalDupCountSql, |
| emptyMap, |
| None, |
| cache = true) |
| finalDupCountTransStep.parentSteps += groupTransStep |
| |
| (finalDupCountTransStep :: targetDsUpdateWriteStep :: Nil, finalDupCountTableName) |
| case _ => |
| (selfGroupTransStep :: Nil, selfGroupTableName) |
| } |
| |
| // 8. distinct metric |
| val distTableName = "__distMetric" |
| val distColName = details.getStringOrKey(_distinct) |
| val distSql = { |
| s""" |
| |SELECT COUNT(*) AS `$distColName` |
| |FROM `$dupCountTableName` WHERE `${ConstantColumns.distinct}` |
| """.stripMargin |
| } |
| val distMetricWriteStep = { |
| MetricWriteStep(distColName, distTableName, EntriesFlattenType, writeTimestampOpt) |
| } |
| val distTransStep = |
| SparkSqlTransformStep(distTableName, distSql, emptyMap, Some(distMetricWriteStep)) |
| |
| val transSteps3 = distTransStep :: Nil |
| |
| val duplicationArrayName = details.getString(_duplicationArray, "") |
| val transSteps4 = if (duplicationArrayName.nonEmpty) { |
| val recordEnable = details.getBoolean(_recordEnable, defValue = false) |
| if (groupAliases.nonEmpty) { |
| // with some group by requirement |
| // 9. origin data join with distinct information |
| val informedTableName = "__informed" |
| val onClause = distAliases |
| .map { alias => |
| s"coalesce(`$sourceAliasTableName`.`$alias`, '') = coalesce(`$dupCountTableName`.`$alias`, '')" |
| } |
| .mkString(" AND ") |
| val informedSql = { |
| s""" |
| |SELECT `$sourceAliasTableName`.*, |
| |`$dupCountTableName`.`$dupColName` AS `$dupColName`, |
| |`$dupCountTableName`.`${ConstantColumns.distinct}` AS `${ConstantColumns.distinct}` |
| |FROM `$sourceAliasTableName` LEFT JOIN `$dupCountTableName` |
| |ON $onClause |
| """.stripMargin |
| } |
| val informedTransStep = SparkSqlTransformStep(informedTableName, informedSql, emptyMap) |
| |
| // 10. add row number |
| val rnTableName = "__rowNumber" |
| val rnDistClause = distAliasesClause |
| val rnSortClause = s"SORT BY `${ConstantColumns.distinct}`" |
| val rnSql = { |
| s""" |
| |SELECT *, |
| |ROW_NUMBER() OVER (DISTRIBUTE BY $rnDistClause $rnSortClause) `${ConstantColumns.rowNumber}` |
| |FROM `$informedTableName` |
| """.stripMargin |
| } |
| val rnTransStep = SparkSqlTransformStep(rnTableName, rnSql, emptyMap) |
| rnTransStep.parentSteps += informedTransStep |
| |
| // 11. recognize duplicate items |
| val dupItemsTableName = "__dupItems" |
| val dupItemsSql = { |
| s""" |
| |SELECT $allAliasesClause, `$dupColName` FROM `$rnTableName` |
| |WHERE NOT `${ConstantColumns.distinct}` OR `${ConstantColumns.rowNumber}` > 1 |
| """.stripMargin |
| } |
| val dupItemsWriteStep = { |
| val rwName = ruleParam |
| .getOutputOpt(RecordOutputType) |
| .flatMap(_.getNameOpt) |
| .getOrElse(dupItemsTableName) |
| RecordWriteStep(rwName, dupItemsTableName, None, writeTimestampOpt) |
| } |
| val dupItemsTransStep = { |
| if (recordEnable) { |
| SparkSqlTransformStep( |
| dupItemsTableName, |
| dupItemsSql, |
| emptyMap, |
| Some(dupItemsWriteStep)) |
| } else { |
| SparkSqlTransformStep(dupItemsTableName, dupItemsSql, emptyMap) |
| } |
| } |
| dupItemsTransStep.parentSteps += rnTransStep |
| |
| // 12. group by dup Record metric |
| val groupDupMetricTableName = "__groupDupMetric" |
| val numColName = details.getStringOrKey(_num) |
| val groupSelClause = groupAliasesClause |
| val groupDupMetricSql = { |
| s""" |
| |SELECT $groupSelClause, `$dupColName`, COUNT(*) AS `$numColName` |
| |FROM `$dupItemsTableName` GROUP BY $groupSelClause, `$dupColName` |
| """.stripMargin |
| } |
| val groupDupMetricWriteStep = { |
| MetricWriteStep( |
| duplicationArrayName, |
| groupDupMetricTableName, |
| ArrayFlattenType, |
| writeTimestampOpt) |
| } |
| val groupDupMetricTransStep = |
| SparkSqlTransformStep( |
| groupDupMetricTableName, |
| groupDupMetricSql, |
| emptyMap, |
| Some(groupDupMetricWriteStep)) |
| groupDupMetricTransStep.parentSteps += dupItemsTransStep |
| |
| groupDupMetricTransStep :: Nil |
| } else { |
| // no group by requirement |
| // 9. duplicate record |
| val dupRecordTableName = "__dupRecords" |
| val dupRecordSelClause = procType match { |
| case StreamingProcessType if withOlderTable => |
| s"$distAliasesClause, `$dupColName`, `$accuDupColName`" |
| |
| case _ => s"$distAliasesClause, `$dupColName`" |
| } |
| val dupRecordSql = { |
| s""" |
| |SELECT $dupRecordSelClause |
| |FROM `$dupCountTableName` WHERE `$dupColName` > 0 |
| """.stripMargin |
| } |
| val dupRecordWriteStep = { |
| val rwName = |
| ruleParam |
| .getOutputOpt(RecordOutputType) |
| .flatMap(_.getNameOpt) |
| .getOrElse(dupRecordTableName) |
| RecordWriteStep(rwName, dupRecordTableName, None, writeTimestampOpt) |
| } |
| val dupRecordTransStep = { |
| if (recordEnable) { |
| SparkSqlTransformStep( |
| dupRecordTableName, |
| dupRecordSql, |
| emptyMap, |
| Some(dupRecordWriteStep), |
| cache = true) |
| } else { |
| SparkSqlTransformStep( |
| dupRecordTableName, |
| dupRecordSql, |
| emptyMap, |
| None, |
| cache = true) |
| } |
| } |
| |
| // 10. duplicate metric |
| val dupMetricTableName = "__dupMetric" |
| val numColName = details.getStringOrKey(_num) |
| val dupMetricSql = { |
| s""" |
| |SELECT `$dupColName`, COUNT(*) AS `$numColName` |
| |FROM `$dupRecordTableName` GROUP BY `$dupColName` |
| """.stripMargin |
| } |
| val dupMetricWriteStep = { |
| MetricWriteStep( |
| duplicationArrayName, |
| dupMetricTableName, |
| ArrayFlattenType, |
| writeTimestampOpt) |
| } |
| val dupMetricTransStep = |
| SparkSqlTransformStep( |
| dupMetricTableName, |
| dupMetricSql, |
| emptyMap, |
| Some(dupMetricWriteStep)) |
| dupMetricTransStep.parentSteps += dupRecordTransStep |
| |
| dupMetricTransStep :: Nil |
| } |
| } else Nil |
| |
| // full steps |
| transSteps1 ++ transSteps2 ++ transSteps3 ++ transSteps4 |
| } |
| } |
| |
| } |
