### Setup FileSystemMetricsRepository

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/metrics_repository_example.md

Initializes a FileSystemMetricsRepository to store metrics on the local disk. Ensure the directory for metrics exists.

```scala
val metricsFile = new File(Files.createTempDir(), "metrics.json")
val repository = FileSystemMetricsRepository(spark, metricsFile.getAbsolutePath)
```

--------------------------------

### Generate Example Data with Scala

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/constraint_suggestion_example.md

Generates sample data using Spark SQL and Scala case classes for demonstrating constraint suggestion.

```scala
case class RawData(
  productName: String,
  totalNumber: String,
  status: String,
  valuable: String
)

val rows = session.sparkContext.parallelize(Seq(
  RawData("thingA", "13.0", "IN_TRANSIT", "true"),
  RawData("thingA", "5", "DELAYED", "false"),
  RawData("thingB", null, "DELAYED", null),
  RawData("thingC", null, "IN_TRANSIT", "false"),
  RawData("thingD", "1.0", "DELAYED", "true"),
  RawData("thingC", "7.0", "UNKNOWN", null),
  RawData("thingC", "24", "UNKNOWN", null),
  RawData("thingE", "20", "DELAYED", "false"),
  RawData("thingA", "13.0", "IN_TRANSIT", "true"),
  RawData("thingA", "5", "DELAYED", "false"),
  RawData("thingB", null, "DELAYED", null),
  RawData("thingC", null, "IN_TRANSIT", "false"),
  RawData("thingD", "1.0", "DELAYED", "true"),
  RawData("thingC", "17.0", "UNKNOWN", null),
  RawData("thingC", "22", "UNKNOWN", null),
  RawData("thingE", "23", "DELAYED", "false")
))

val data = session.createDataFrame(rows)
```

--------------------------------

### Example Constraint Suggestion for 'valuable' column

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/constraint_suggestion_example.md

Illustrates Deequ's suggestions for the 'valuable' column, identifying it as a boolean type and suggesting a completeness constraint based on observed missing values.

```text
Constraint suggestion for 'valuable': 'valuable' has type Boolean
The corresponding scala code is .hasDataType("valuable", ConstrainableDataTypes.Boolean)

Constraint suggestion for 'valuable': 'valuable' has less than 62% missing values
The corresponding scala code is .hasCompleteness("valuable", _ >= 0.38, Some("It should be above 0.38!"))
```

--------------------------------

### Example Constraint Suggestion for 'totalNumber' column

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/constraint_suggestion_example.md

Shows Deequ's suggestions for the 'totalNumber' column, recognizing it as a numeric type, suggesting a completeness constraint, and inferring a positivity constraint based on observed data.

```text
Constraint suggestion for 'totalNumber': 'totalNumber' has type Fractional
The corresponding scala code is .hasDataType("totalNumber", ConstrainableDataTypes.Fractional)

Constraint suggestion for 'totalNumber': 'totalNumber' has less than 47% missing values
The corresponding scala code is .hasCompleteness("totalNumber", _ >= 0.53, Some("It should be above 0.53!"))

```

--------------------------------

### Metrics Repository Content Example

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/anomaly_detection_example.md

This table displays the content of the metrics repository, showing the 'Size' metric for two different timestamps, illustrating the data size increase from 2.0 to 5.0.

```text
+-------+--------+----+-----+-------------+
| entity|instance|Name|value| dataset_date|
+-------+--------+----+-----+-------------+
|Dataset|       *|Size|  2.0|1538384009558|
|Dataset|       *|Size|  5.0|1538385453983|
+-------+--------+----+-----+-------------+

```

--------------------------------

### Run Default Constraint Suggestion with Deequ

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/constraint_suggestion_example.md

Computes constraint suggestions using Deequ's default set of heuristic rules. This is a common starting point for constraint discovery.

```scala
val suggestionResult = ConstraintSuggestionRunner()
  .onData(data)
  .addConstraintRules(Rules.DEFAULT)
  .run()
```

--------------------------------

### Java Example: Evaluate Data Quality with DQDL Rules

Source: https://github.com/awslabs/deequ/blob/master/README.md

This snippet demonstrates how to define and evaluate data quality rules using DQDL syntax in Java. It requires SparkSession and Deequ's EvaluateDataQuality. Use this for dataset-level checks.

```java
import com.amazon.deequ.dqdl.EvaluateDataQuality;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

SparkSession spark = SparkSession.builder()
    .appName("DQDL Java Example")
    .master("local[*]")
    .getOrCreate();

// Create sample data
Dataset<Row> df = spark.sql(
    "SELECT * FROM VALUES " +
    "('1', 'a', 'c'), " +
    "('2', 'a', 'c'), " +
    "('3', 'a', 'c'), " +
    "('4', 'b', 'd') " +
    "AS t(item, att1, att2)"
);

// Define rules using DQDL syntax
String ruleset = "Rules=[IsUnique \"item\", RowCount < 10, Completeness \"item\" > 0.8, Uniqueness \"item\" = 1.0]";

// Evaluate data quality
Dataset<Row> results = EvaluateDataQuality.process(df, ruleset);
results.show()
```

--------------------------------

### Deequ Composite Rules Example

Source: https://github.com/awslabs/deequ/blob/master/README.md

Demonstrates various ways to define and use composite rules in Deequ, including simple AND/OR, nested logic, and multiple rules. Ensure SparkSession and DataFrame are set up before running.

```scala
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()

// Multiple composite rules in one ruleset
val multipleRules = """Rules=[
  (RowCount > 0) and (IsComplete "id"),
  (IsUnique "id") or (IsUnique "email"),
  ((Mean "age" > 20) and (Mean "age" < 50)) or (RowCount < 10)
]"""
val multipleResults = EvaluateDataQuality.process(df, multipleRules)
multipleResults.show()
```

--------------------------------

### Example Output of Data Quality Checks

Source: https://github.com/awslabs/deequ/blob/master/README.md

This output shows the results of data quality checks, highlighting violations of completeness and pattern constraints. It indicates the percentage of values that did not meet the specified requirements.

```text
We found errors in the data:

CompletenessConstraint(Completeness(productName)): Value: 0.8 does not meet the requirement!
PatternConstraint(containsURL(description)): Value: 0.4 does not meet the requirement!
```

--------------------------------

### Scala Example: Evaluate Data Quality with DQDL Rules

Source: https://github.com/awslabs/deequ/blob/master/README.md

This snippet demonstrates how to define and evaluate data quality rules using DQDL syntax in Scala. It requires SparkSession and Deequ's EvaluateDataQuality. Use this for dataset-level checks.

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

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

import spark.implicits._

// Sample data
val df = Seq(
  ("1", "a", "c"),
  ("2", "a", "c"),
  ("3", "a", "c"),
  ("4", "b", "d")
).toDF("item", "att1", "att2")

// Define rules using DQDL syntax
val ruleset = """Rules=[IsUnique "item", RowCount < 10, Completeness "item" > 0.8, Uniqueness "item" = 1.0]"""

// Evaluate data quality
val results = EvaluateDataQuality.process(df, ruleset)
results.show()
```

--------------------------------

### Define Item Data Structure

Source: https://github.com/awslabs/deequ/blob/master/README.md

Defines the case class for the 'Item' data structure used in the example. This structure represents the schema of the data to be analyzed.

```scala
case class Item(
  id: Long,
  productName: String,
  description: String,
  priority: String,
  numViews: Long
)
```

--------------------------------

### Scala Example: Row-Level Results with DQDL Rules

Source: https://github.com/awslabs/deequ/blob/master/README.md

This snippet shows how to obtain row-level data quality results using DQDL in Scala. It uses `processRows` and is suitable for rules that can be evaluated per row, such as `IsComplete` or `ColumnValues`.

```scala
val df = Seq(
  ("1", "Alice", Some(25)),
  ("2", "Bob", None),
  ("3", null, Some(30))
).toDF("id", "name", "age")

val ruleset = """Rules=[IsComplete "name", IsComplete "age"]"""

val results = EvaluateDataQuality.processRows(df, ruleset)

// Access the row-level outcomes
val rowLevelOutcomes = results("rowLevelOutcomes")
rowLevelOutcomes.select("id", "DataQualityRulesPass", "DataQualityRulesFail", "DataQualityEvaluationResult").show(false)

// Filter to failed rows
val failedRows = rowLevelOutcomes.filter($"DataQualityEvaluationResult" === "Failed")
```

--------------------------------

### Create Toy Dataframe

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/metrics_repository_example.md

Sets up a sample Spark DataFrame with item data for metric computation.

```scala
val data = ExampleUtils.itemsAsDataframe(spark,
  Item(1, "Thingy A", "awesome thing.", "high", 0),
  Item(2, "Thingy B", "available at http://thingb.com", null, 0),
  Item(3, null, null, "low", 5),
  Item(4, "Thingy D", "checkout https://thingd.ca", "low", 10),
  Item(5, "Thingy E", null, "high", 12))
```

--------------------------------

### Run Initial Analysis with State Saving

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

Execute the defined analysis on the initial data and configure a StateProvider to save the internal computation states. This allows for future incremental updates.

```scala
val stateStore = InMemoryStateProvider()

val metricsForData = AnalysisRunner.run(
  data = data,
  analysis = analysis,
  saveStatesWith = Some(stateStore))
```

--------------------------------

### Load Metrics as DataFrame by Tag

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/metrics_repository_example.md

Queries the repository for metrics tagged with 'repositoryExample' and returns the results as a Spark DataFrame.

```scala
repository.load()
  .withTagValues(Map("tag" -> "repositoryExample"))
  .getSuccessMetricsAsDataFrame(spark)
  .show()
```

--------------------------------

### Output of Initial Metrics

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

The printed output shows the computed metrics for the initial dataset, including size, distinct count, and completeness.

```text
Size(None): 3.0
ApproxCountDistinct(id,None): 3.0
Completeness(productName,None): 1.0
Completeness(description,None): 0.6666666666666666
```

--------------------------------

### Define Analysis and Initial Data

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

Define the metrics to compute and generate the initial dataset. This sets up the analyzers and the first batch of data for analysis.

```scala
val data = ExampleUtils.itemsAsDataframe(spark,
  Item(1, "Thingy A", "awesome thing.", "high", 0),
  Item(2, "Thingy B", "available tomorrow", "low", 0),
  Item(3, "Thing C", null, null, 5))

val analysis = Analysis()
  .addAnalyzer(Size())
  .addAnalyzer(ApproxCountDistinct("id"))
  .addAnalyzer(Completeness("productName"))
  .addAnalyzer(Completeness("description"))
```

--------------------------------

### Create Sample Spark DataFrame

Source: https://github.com/awslabs/deequ/blob/master/README.md

Generates a Spark DataFrame with sample 'Item' data. This DataFrame is used as input for Deequ's data quality checks.

```scala
val rdd = spark.sparkContext.parallelize(Seq(
  Item(1, "Thingy A", "awesome thing.", "high", 0),
  Item(2, "Thingy B", "available at http://thingb.com", null, 0),
  Item(3, null, null, "low", 5),
  Item(4, "Thingy D", "checkout https://thingd.ca", "low", 10),
  Item(5, "Thingy E", null, "high", 12)))

val data = spark.createDataFrame(rdd)
```

--------------------------------

### Initialize Metrics Repository

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/anomaly_detection_example.md

An in-memory metrics repository is created to store and retrieve metric data for anomaly detection.

```scala
val metricsRepository = new InMemoryMetricsRepository()
```

--------------------------------

### Load Specific Metric by Key

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/metrics_repository_example.md

Retrieves a specific metric (e.g., Completeness of 'productName') from the repository using the ResultKey.

```scala
val completenessOfProductName = repository
  .loadByKey(resultKey).get
  .metric(Completeness("productName")).get

println(s"The completeness of the productName column is: $completenessOfProductName")
```

--------------------------------

### Inspect Initial Metrics

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

Print the computed metrics from the initial analysis. This shows the data quality metrics for the first set of records.

```scala
println("Metrics for the first 3 records:\n")
metricsForData.metricMap.foreach { case (analyzer, metric) =>
  println(s"\t$analyzer: ${metric.value.get}")
}
```

--------------------------------

### Investigate Suggested Constraints

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/constraint_suggestion_example.md

Iterates through the computed constraint suggestions and prints their textual description and the corresponding Scala code. Manual review of these suggestions is recommended.

```scala
suggestionResult.constraintSuggestions.foreach { case (column, suggestions) =>
  suggestions.foreach {
    case suggestion =>
      println(s"Constraint suggestion for '$column':\t${suggestion.description}\n" +
        s"The corresponding scala code is ${suggestion.codeForConstraint}\n")
  }
}
```

--------------------------------

### Run Verification and Save Metrics

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/metrics_repository_example.md

Executes verification checks on the data and saves the resulting metrics to the configured repository using the specified ResultKey.

```scala
VerificationSuite()
  .onData(data)
  .addCheck(Check(CheckLevel.Error, "integrity checks")
    .hasSize(_ == 5)
    .isComplete("id")
    .isComplete("productName")
    .isContainedIn("priority", Array("high", "low"))
    .isNonNegative("numViews"))
  .useRepository(repository)
  .saveOrAppendResult(resultKey)
  .run()
```

--------------------------------

### Define Yesterday's Dataset

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/anomaly_detection_example.md

Create a sample DataFrame representing yesterday's data. This is used to establish a baseline for anomaly detection.

```scala
val yesterdaysDataset = itemsAsDataframe(session,
  Item(1, "Thingy A", "awesome thing.", "high", 0),
  Item(2, "Thingy B", "available at http://thingb.com", null, 0))

```

--------------------------------

### Run Custom Constraint Suggestion with Deequ

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/constraint_suggestion_example.md

Allows for customizing constraint suggestions by adding individual constraint rules, such as the RetainCompletenessRule. Use this when specific rules are of interest.

```scala
val suggestionResult = ConstraintSuggestionRunner()
  .onData(data)
  
  .addConstraintRule(
    RetainCompletenessRule(intervalStrategy = WilsonScoreIntervalStrategy())
  )
  .run()
```

--------------------------------

### Define and Run Deequ Data Quality Checks

Source: https://github.com/awslabs/deequ/blob/master/README.md

Sets up and executes a series of data quality checks on the Spark DataFrame using Deequ's VerificationSuite. This includes checks for size, completeness, uniqueness, value containment, non-negativity, URL presence, and approximate quantiles.

```scala
import com.amazon.deequ.VerificationSuite
import com.amazon.deequ.checks.{Check, CheckLevel, CheckStatus}


val verificationResult = VerificationSuite()
  .onData(data)
  .addCheck(
    Check(CheckLevel.Error, "unit testing my data")
      .hasSize(_ == 5) // we expect 5 rows
      .isComplete("id") // should never be NULL
      .isUnique("id") // should not contain duplicates
      .isComplete("productName") // should never be NULL
      // should only contain the values "high" and "low"
      .isContainedIn("priority", Array("high", "low"))
      .isNonNegative("numViews") // should not contain negative values
      // at least half of the descriptions should contain a url
      .containsURL("description", _ >= 0.5)
      // half of the items should have less than 10 views
      .hasApproxQuantile("numViews", 0.5, _ <= 10))
    .run()
```

--------------------------------

### Load Partitioned DataFrames

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

Loads manufacturer data into DataFrames, partitioned by country code (DE, US, CN). This sets up the data for stateful analysis.

```scala
val deManufacturers = ExampleUtils.manufacturersAsDataframe(spark,
  Manufacturer(1, "ManufacturerA", "DE"),
  Manufacturer(2, "ManufacturerB", "DE"))

val usManufacturers = ExampleUtils.manufacturersAsDataframe(spark,
  Manufacturer(3, "ManufacturerD", "US"),
  Manufacturer(4, "ManufacturerE", "US"),
  Manufacturer(5, "ManufacturerF", "US"))

val cnManufacturers = ExampleUtils.manufacturersAsDataframe(spark,
  Manufacturer(6, "ManufacturerG", "CN"),
  Manufacturer(7, "ManufacturerH", "CN"))
```

--------------------------------

### Define Today's Dataset

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/anomaly_detection_example.md

Create a sample DataFrame representing today's data, which has more rows than yesterday's to potentially trigger an anomaly.

```scala
val todaysDataset = itemsAsDataframe(session,
  Item(1, "Thingy A", "awesome thing.", "high", 0),
  Item(2, "Thingy B", "available at http://thingb.com", null, 0),
  Item(3, null, null, "low", 5),
  Item(4, "Thingy D", "checkout https://thingd.ca", "low", 10),
  Item(5, "Thingy E", null, "high", 12))

```

--------------------------------

### Suggest Completeness Constraint for 'productName'

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/constraint_suggestion_example.md

Use `.isComplete()` to suggest a constraint that checks if a column has any missing values. This is useful when a column should not contain nulls.

```scala
.isComplete("productName")
```

--------------------------------

### Run Anomaly Check for Yesterday

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/anomaly_detection_example.md

Configure and run a VerificationSuite to test for anomalies in the dataset size using a RateOfChangeStrategy. Metrics are saved to the repository with a timestamp.

```scala
val yesterdaysKey = ResultKey(System.currentTimeMillis() - 24 * 60 * 60 * 1000)

VerificationSuite()
  .onData(yesterdaysDataset)
  .useRepository(metricsRepository)
  .saveOrAppendResult(yesterdaysKey)
  .addAnomalyCheck(
    RelativeRateOfChangeStrategy(maxRateIncrease = Some(2.0)),
    Size())
  .run()
```

--------------------------------

### Load All Metrics as JSON

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/metrics_repository_example.md

Fetches all metrics stored in the repository within the last 10 minutes and returns them as a JSON string.

```scala
val json = repository.load()
  .after(System.currentTimeMillis() - 10000)
  .getSuccessMetricsAsJson()

println(json)
```

--------------------------------

### Run Column Profiling with Deequ

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/data_profiling_example.md

Initiate the column profiling process on a Spark DataFrame using Deequ's ColumnProfilerRunner. This method performs multiple passes over the data efficiently.

```scala
val result = ColumnProfilerRunner()
  .onData(rawData)
  .run()
```

--------------------------------

### Suggest Completeness Constraint for 'status'

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/constraint_suggestion_example.md

Use `.isComplete()` to suggest a constraint that checks for the absence of missing values in the 'status' column.

```scala
.isComplete("status")
```

--------------------------------

### Run Anomaly Check for Today

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/anomaly_detection_example.md

Execute the VerificationSuite again with today's data, saving the results to the repository. This check is expected to detect an anomaly due to the significant increase in data size.

```scala
val todaysKey = ResultKey(System.currentTimeMillis())

val verificationResult = VerificationSuite()
  .onData(todaysDataset)
  .useRepository(metricsRepository)
  .saveOrAppendResult(todaysKey)
  .addAnomalyCheck(
    RelativeRateOfChangeStrategy(maxRateIncrease = Some(2.0)),
    Size())
  .run()

```

--------------------------------

### Update Partition Data and Recompute States

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

Updates the US partition data with new values, including nulls and URLs, and recomputes only the states for this changed partition. This demonstrates efficient updates.

```scala
val updatedUsManufacturers = ExampleUtils.manufacturersAsDataframe(spark,
  Manufacturer(3, "ManufacturerDNew", "US"),
  Manufacturer(4, null, "US"),
  Manufacturer(5, "ManufacturerFNew http://clickme.com", "US"))

val updatedUsStates = InMemoryStateProvider()

AnalysisRunner.run(
  updatedUsManufacturers,
  analysis,
  saveStatesWith = Some(updatedUsStates)
)
```

--------------------------------

### Compute and Store Partition States

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

Computes and stores the states for the defined analysis on each partition (DE, US, CN) using InMemoryStateProvider. This prepares for aggregated state computation.

```scala
val analysis = Analysis(check.requiredAnalyzers().toSeq)

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))
```

--------------------------------

### Analyze Value Distribution for Categorical Columns

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/data_profiling_example.md

Retrieve and display the value distribution for columns with a limited number of distinct values. This includes the count and ratio of each unique value present in the column.

```scala
val statusProfile = result.profiles("status")

println("Value distribution in 'status':")
statusProfile.histogram.foreach {
  _.values.foreach { case (key, entry) =>
    println(s"\t$key occurred ${entry.absolute} times (ratio is ${entry.ratio})")
  }
}
```

--------------------------------

### Define Raw Data Case Class and Spark DataFrame

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/data_profiling_example.md

Define a case class for raw data and create a Spark DataFrame from a collection of these case class instances. This sets up the data structure for profiling.

```scala
case class RawData(productName: String, totalNumber: String, status: String, valuable: String)

val rows = session.sparkContext.parallelize(Seq(
  RawData("thingA", "13.0", "IN_TRANSIT", "true"),
  RawData("thingA", "5", "DELAYED", "false"),
  RawData("thingB", null,  "DELAYED", null),
  RawData("thingC", null, "IN_TRANSIT", "false"),
  RawData("thingD", "1.0",  "DELAYED", "true"),
  RawData("thingC", "7.0", "UNKNOWN", null),
  RawData("thingC", "20", "UNKNOWN", null),
  RawData("thingE", "20", "DELAYED", "false")
))

val rawData = session.createDataFrame(rows)
```

--------------------------------

### Define ResultKey for Metrics

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/metrics_repository_example.md

Creates a ResultKey which includes a timestamp and arbitrary tags to index computed metrics.

```scala
val resultKey = ResultKey(
  System.currentTimeMillis(),
  Map("tag" -> "repositoryExample"))
```

--------------------------------

### Define Metrics Check

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

Defines a Deequ Check with warning level, specifying metrics for completeness, pattern matching, and compliance on the manufacturer data.

```scala
val check = Check(CheckLevel.Warning, "a check")
  .isComplete("manufacturerName")
  .containsURL("manufacturerName", _ == 0.0)
  .isContainedIn("countryCode", Array("DE", "US", "CN"))
```

--------------------------------

### Check Anomaly Detection Status and Display Metrics

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/anomaly_detection_example.md

Inspect the verification result status. If an anomaly is detected, print a message and display the metrics stored in the repository, showing the historical and current values.

```scala
if (verificationResult.status != Success) {
  println("Anomaly detected in the Size() metric!")

  metricsRepository
    .load()
    .forAnalyzers(Seq(Size()))
    .getSuccessMetricsAsDataFrame(session)
    .show()
}

```

--------------------------------

### Compute Table Metrics from Aggregated States

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

Calculates the metrics for the entire table by running AnalysisRunner.runOnAggregatedStates with the collected states from all partitions. This avoids re-processing the raw data.

```scala
val tableMetrics = AnalysisRunner.runOnAggregatedStates(
  deManufacturers.schema,
  analysis,
  Seq(deStates, usStates, cnStates)
)

println("Metrics for the whole table:\n")
tableMetrics.metricMap.foreach { case (analyzer, metric) =>
  println(s"\t$analyzer: ${metric.value.get}")
}
```

--------------------------------

### Run Incremental Analysis with Stored States

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

Process new data by continuing from the previously saved internal state. This updates the metrics for the combined dataset without re-reading the old data.

```scala
val moreData = ExampleUtils.itemsAsDataframe(spark,
  Item(4, "Thingy D", null, "low", 10),
  Item(5, "Thingy E", null, "high", 12))

val metricsAfterAddingMoreData = AnalysisRunner.run(
  data = moreData,
  analysis = analysis,
  aggregateWith = Some(stateStore)
)

println("\nMetrics after adding 2 more records:\n")
metricsAfterAddingMoreData.metricMap.foreach { case (analyzer, metric) =>
  println(s"\t$analyzer: ${metric.value.get}")
}
```

--------------------------------

### sbt Dependency for Deequ (Spark 3.1)

Source: https://github.com/awslabs/deequ/blob/master/README.md

Add this sbt dependency to your project to include Deequ for Spark 3.1.x. Ensure your Spark version is compatible.

```scala
libraryDependencies += "com.amazon.deequ" % "deequ" % "2.0.0-spark-3.1"
```

--------------------------------

### Suggest Value Range Constraint for 'productName'

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/constraint_suggestion_example.md

Use `.isContainedIn()` to suggest a constraint that verifies if all values in a column are present within a predefined array of acceptable values. This is suitable for columns with a limited set of possible entries.

```scala
.isContainedIn("productName", Array("thingC", "thingA", "thingB", "thingE", "thingD"))
```

--------------------------------

### Compute Updated Table Metrics

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

Calculates the updated metrics for the whole table using the existing states for DE and CN, and the newly computed state for the updated US partition. This highlights the efficiency of stateful computation.

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

println("Metrics for the whole table after updating the US partition:\n")
updatedTableMetrics.metricMap.foreach { case (analyzer, metric) =>
  println(s"\t$analyzer: ${metric.value.get}")
}
```

--------------------------------

### Maven Dependency for Deequ (Spark 3.1)

Source: https://github.com/awslabs/deequ/blob/master/README.md

Add this Maven dependency to your project to include Deequ for Spark 3.1.x. Ensure your Spark version is compatible.

```xml
<dependency>
  <groupId>com.amazon.deequ</groupId>
  <artifactId>deequ</artifactId>
  <version>2.0.0-spark-3.1</version>
</dependency>
```

--------------------------------

### Output of Updated Metrics

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md

The printed output shows the updated data quality metrics after incorporating the new records, reflecting the combined dataset.

```text
Size(None): 5.0
ApproxCountDistinct(id,None): 5.0
Completeness(productName,None): 1.0
Completeness(description,None): 0.4
```

--------------------------------

### Inspect Column Profiling Results

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/data_profiling_example.md

Iterate through the profiling results for each column to access metrics like completeness, approximate number of distinct values, and inferred data type. This provides a high-level overview of the data.

```scala
result.profiles.foreach { case (colName, profile) =>
  println(s"Column '$colName':\n " +
    s"\tcompleteness: ${profile.completeness}\n" +
    s"\tapproximate number of distinct values: ${profile.approximateNumDistinctValues}\n" +
    s"\tdatatype: ${profile.dataType}\n")
}
```

--------------------------------

### Access Numeric Column Statistics

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/data_profiling_example.md

For columns identified as numeric, cast the profile to NumericColumnProfile to access detailed statistics such as minimum, maximum, mean, and standard deviation. Ensure the column is indeed numeric before casting.

```scala
val totalNumberProfile = result.profiles("totalNumber").asInstanceOf[NumericColumnProfile]

println(s"Statistics of 'totalNumber':\n" +
  s"\tminimum: ${totalNumberProfile.minimum.get}\n" +
  s"\tmaximum: ${totalNumberProfile.maximum.get}\n" +
  s"\tmean: ${totalNumberProfile.mean.get}\n" +
  s"\tstandard deviation: ${totalNumberProfile.stdDev.get}\n")
```

--------------------------------

### Suggest Non-Negative Constraint for 'totalNumber'

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/constraint_suggestion_example.md

Use `.isNonNegative()` to suggest a constraint that checks for the absence of negative values in a specified column.

```scala
.isNonNegative("totalNumber")
```

--------------------------------

### Suggest Value Range Constraint for 'status'

Source: https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/constraint_suggestion_example.md

Use `.isContainedIn()` to suggest a constraint that ensures all 'status' values fall within the specified array of known statuses.

```scala
.isContainedIn("status", Array("DELAYED", "UNKNOWN", "IN_TRANSIT"))
```

--------------------------------

### Inspect Deequ Verification Results

Source: https://github.com/awslabs/deequ/blob/master/README.md

Analyzes the VerificationResult object returned by Deequ to determine if the data passed all defined checks. It prints a success message or lists the specific constraints that failed with their corresponding messages.

```scala
import com.amazon.deequ.constraints.ConstraintStatus


if (verificationResult.status == CheckStatus.Success) {
  println("The data passed the test, everything is fine!")
} else {
  println("We found errors in the data:\n")

  val resultsForAllConstraints = verificationResult.checkResults
    .flatMap { case (_, checkResult) => checkResult.constraintResults }

  resultsForAllConstraints
    .filter { _.status != ConstraintStatus.Success }
    .foreach { result => println(s"${result.constraint}: ${result.message.get}") }
}
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.