Deequ Documentation

To avoid re-reading entire datasets when new data is appended, use Deequ's stateful metrics computation. This process involves two steps:

1. Initial Run: Run an Analysis using AnalysisRunner.run and provide a StateProvider (e.g., InMemoryStateProvider) via the saveStatesWith parameter to capture the internal computation state.

2. Incremental Update: When new data arrives, run AnalysisRunner.run on only the new data, but use the aggregateWith parameter to pass the previously stored StateProvider. This allows Deequ to update the metrics for the entire dataset without accessing the old data.

// 1. Initial run with state saving
val stateStore = InMemoryStateProvider()
val metricsForData = AnalysisRunner.run(
  data = data,
  analysis = analysis,
  saveStatesWith = Some(stateStore)
)

// 2. Incremental update using the stored state
val metricsAfterAddingMoreData = AnalysisRunner.run(
  data = moreData,
  analysis = analysis,
  aggregateWith = Some(stateStore)
)

Deequ supports anomaly detection by comparing current data quality metrics against historical values stored in a MetricsRepository. This allows you to detect unexpected changes in metrics (like dataset size) without defining hard constraints.

To implement anomaly detection:

1. Initialize a MetricsRepository (e.g., InMemoryMetricsRepository).
2. Store historical metrics using .useRepository(metricsRepository) and .saveOrAppendResult(resultKey) within a VerificationSuite.
3. Define an anomaly check using .addAnomalyCheck(strategy, metric).
4. Use a strategy like RelativeRateOfChangeStrategy to define the threshold for what constitutes an anomaly.

import com.amazon.deequ.VerificationSuite
import com.amazon.deequ.repository.InMemoryMetricsRepository
import com.amazon.deequ.implicits._
import com.amazon.deequ.anomalydetection.RelativeRateOfChangeStrategy
import com.amazon.deequ.checks.CheckLevel
import com.amazon.deequ.checks.CheckStatus

// 1. Setup repository
val metricsRepository = new InMemoryMetricsRepository()

// 2. Define a key for historical data (e.g., yesterday)
val yesterdaysKey = ResultKey(System.currentTimeMillis() - 24 * 60 * 60 * 1000)

// 3. Run verification and save to repository
VerificationSuite()
  .onData(yesterdaysDataset)
  .useRepository(metricsRepository)
  .saveOrAppendResult(yesterdaysKey)
  .addAnomalyCheck(
    RelativeRateOfChangeStrategy(maxRateIncrease = Some(2.0)),
    Size()
  )
  .run()

// 4. Run anomaly check on current data
val todaysKey = ResultKey(System.currentTimeMillis())
val verificationResult = VerificationSuite()
  .onData(todaysDataset)
  .useRepository(metricsRepository)
  .saveOrAppendResult(todaysKey)
  .addAnomalyCheck(
    RelativeRateOfChangeStrategy(maxRateIncrease = Some(2.0)),
    Size()
  )
  .run()

// 5. Inspect results
if (verificationResult.status != Success) {
  println("Anomaly detected!")
  metricsRepository
    .load()
    .forAnalyzers(Seq(Size()))
    .getSuccessMetricsAsDataFrame(session)
    .show()
}

You can combine multiple DQDL rules using and and or operators to create complex validation logic.

Note: Composite rules currently only support dataset-level evaluation. Row-level evaluation for composite rules is not yet implemented.

import com.amazon.deequ.dqdl.EvaluateDataQuality
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("Composite Rules Example")
  .master("local[*]")
  .getOrCreate()

import spark.implicits._

val df = Seq(
  (1, "Alice", 25, "alice@example.com"),
  (2, "Bob", 30, "bob@example.com"),
  (3, "Charlie", 35, "charlie@example.com")
).toDF("id", "name", "age", "email")

// Simple AND: Both conditions must be true
val andRule = """Rules=[(RowCount > 0) and (IsComplete "email")]"""
val andResults = EvaluateDataQuality.process(df, andRule)
andResults.show()

// Simple OR: At least one condition must be true
val orRule = """Rules=[(RowCount > 100) or (IsUnique "id")]"""
val orResults = EvaluateDataQuality.process(df, orRule)
orResults.show()

// Nested composition: Complex logic with multiple levels
val nestedRule = """Rules=[
  ((IsComplete "name") and (IsComplete "email")) or 
  ((RowCount > 0) and (IsUnique "id"))
]"""
val nestedResults = EvaluateDataQuality.process(df, nestedRule)
nestedResults.show()

If your data is partitioned (e.g., by country code), you can compute metrics for the whole table by aggregating the states of individual partitions. This is highly efficient when a single partition changes, as you only need to recompute the state for that specific partition and then re-aggregate.

1. Compute Partition States: Run AnalysisRunner.run for each partition, saving the state of each using a StateProvider via saveStatesWith.

2. Aggregate for Whole Table: Use AnalysisRunner.runOnAggregatedStates to compute the final metrics for the entire table by passing the schema, the analysis, and a sequence of the partition StateProviders.

3. Handle Updates: If a partition changes, re-run the analysis on only that partition to get a new StateProvider, then call runOnAggregatedStates again with the updated sequence of states.

// 1. Compute and save states for each partition
val deStates = InMemoryStateProvider()
val usStates = InMemoryStateProvider()
val cnStates = InMemoryStateProvider()

AnalysisRunner.run(deManufacturers, analysis, saveStatesWith = Some(deStates))
AnalysisRunner.run(usManufacturers, analysis, saveStatesWith = Some(usStates))
AnalysisRunner.run(cnManufacturers, analysis, saveStatesWith = Some(cnStates))

// 2. Aggregate states to get metrics for the whole table
val tableMetrics = AnalysisRunner.runOnAggregatedStates(
  deManufacturers.schema,
  analysis,
  Seq(deStates, usStates, cnStates)
)

// 3. Update only one partition and re-aggregate
val updatedUsStates = InMemoryStateProvider()
AnalysisRunner.run(updatedUsManufacturers, analysis, saveStatesWith = Some(updatedUsStates))

val updatedTableMetrics = AnalysisRunner.runOnAggregatedStates(
  deManufacturers.schema,
  analysis,
  Seq(deStates, updatedUsStates, cnStates)
)

Deequ can automatically suggest data quality constraints by profiling your data and applying heuristic rules. This helps identify appropriate data types (even if 'disguised' as strings), completeness requirements, and value ranges.

To use this feature, use the ConstraintSuggestionRunner to specify your data, add constraint rules, and run the process. The result provides a textual description and the corresponding Scala code for each suggested constraint.

Note: Suggestions are based on heuristics and assume the provided data is 'static' and correct. Always manually review suggestions before applying them to production deployments.

// 1. Initialize the runner on your DataFrame
val suggestionResult = ConstraintSuggestionRunner()
  .onData(data)
  .addConstraintRules(Rules.DEFAULT)
  .run()

// 2. Iterate through suggestions to see descriptions and Scala code
suggestionResult.constraintSuggestions.foreach { case (column, suggestions) =>
  suggestions.foreach { suggestion =>
    println(s"Constraint suggestion for '$column':\t${suggestion.description}\n" +
      s"The corresponding scala code is ${suggestion.codeForConstraint}\n")
  }
}

Deequ allows you to persist computed metrics using a MetricsRepository. You can use FileSystemMetricsRepository to store metrics in JSON format on a local filesystem, HDFS, or S3.

To use a repository:

1. Initialize a FileSystemMetricsRepository with a Spark session and a file path.
2. Create a ResultKey containing a timestamp and optional metadata tags.
3. In your VerificationSuite, call .useRepository(repository) and .saveOrAppendResult(resultKey) to persist the results.

Once stored, you can retrieve metrics by specific analyzers, filter by time ranges, or query by tags.

import com.amazon.deequ.repository.fs.FileSystemMetricsRepository
import com.amazon.deequ.ResultKey
import java.io.File
import java.nio.file.Files

// 1. Setup Repository
val metricsFile = new File(Files.createTempDir(), "metrics.json")
val repository = FileSystemMetricsRepository(spark, metricsFile.getAbsolutePath)

// 2. Define a ResultKey for indexing
val resultKey = ResultKey(
  System.currentTimeMillis(),
  Map("tag" -> "repositoryExample")
)

// 3. Run Verification and save to repository
VerificationSuite()
  .onData(data)
  .addCheck(Check(CheckLevel.Error, "integrity checks")
    .hasSize(_ == 5)
    .isComplete("id")
  )
  .useRepository(repository)
  .saveOrAppendResult(resultKey)
  .run()