| /* |
| 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._ |
| 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, 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, 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, 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, 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, false) |
| if (groupAliases.size > 0) { |
| // 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), |
| true |
| ) |
| } else { |
| SparkSqlTransformStep(dupRecordTableName, dupRecordSql, emptyMap, None, 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 |
| } |
| } |
| |
| } |
