Adding new MetricRepository to read and write data via REST by cmachgodaddy · Pull Request #445 · awslabs/deequ

cmachgodaddy · 2023-01-12T22:08:22Z

Issue #, if available:

Description of changes:

Adding new MetricRepository to read and write data via REST

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Adding new MetricRepository for reading and writing data through REST

cmachgodaddy · 2023-01-12T22:14:56Z

@shehzad-qureshi @eycho-am @sseb @sseb Can you review this PR?

rdsharma26 · 2023-01-25T16:58:22Z

Thank you for the pull request. Can you please share more details about your use case? Can you also attach any testing evidence that showcases how the changes work?

rdsharma26 · 2023-07-05T20:17:54Z

@cmachgodaddy Can you provide an update on my comment above?

cmachgodaddy · 2024-06-13T22:16:53Z

Hi @rdsharma26 , sorry for the late response. My use case was as bellow:
I have a Dynamo DB table, which has a metric table, which contains all of the metric values output from the Deeque so far. And those metrics written by our API to the Dynamo. So, I added this repo as REST, instead of a direct Dynamo ingression.

Another reason it has to be API is our existing metric has its own schema which is not compatible with the schema that Deequ's Anomaly Detection expected. So providing this API as a abstraction layer has the flexibility to convert the schema from one format to another.

I have used this REST repo for a while now, and in production, and it works quite well. Here is how I used it

    val ENDPOINT = apiHelper.get.dlmsURL
    val PATH = s"/v1/accounts/${getAccountId()}/databases/${config.database}/tables/${config.table}/xxx"
    val headers = Map(
      "Content-Type" -> "application/json",
      "Authorization" -> s"sso-jwt xxx"
    )

    val writeRequest = new DefaultRequest[Void]("execute-api")
    writeRequest.setHttpMethod(HttpMethodName.POST)
    writeRequest.setEndpoint(URI.create(ENDPOINT))
    writeRequest.setResourcePath(PATH)
    writeRequest.setHeaders(headers.asJava)

    val readRequest = new DefaultRequest[Void]("execute-api")
    readRequest.setHttpMethod(HttpMethodName.GET)
    readRequest.setEndpoint(URI.create(ENDPOINT))
    readRequest.setResourcePath(PATH)
    readRequest.setHeaders(headers.asJava)
    val repo = new RestMetricsRepository(readRequest, writeRequest)

And, here is how I run the Anomaly Detection job, and send data to Dynamo

      val verificationResult = VerificationSuite()
        .onData(dataDF)
        .useRepository(repo)
        .saveOrAppendResult(todaysKey)
        .addAnomalyCheck(strategyObj, metricObj)
        .run()

These should be in the PR, btw.

I can't snapshot the results in my Dynamo, since it contains some sensitive data. But, the changes in this PR should be very straightforward? it based off the existing patterns of the FileSystem and the InMemory repos.

Let me know if you need anything else to get this PR in?

github-actions

Generated by AI (model: us.anthropic.claude-opus-4-6-v1, prompt: 38d3f74c) — may not be fully accurate. Reply if this doesn't help.

github-actions · 2026-05-06T21:31:16Z


-
+        <dependency>
+            <groupId>com.amazonaws</groupId>


The aws-java-sdk-core dependency should have <scope>provided</scope> or <scope>compile</scope> with an <optional>true</optional> tag. Adding it as a hard compile dependency forces all Deequ users to pull in the AWS SDK even if they don't use RestMetricsRepository. This is inconsistent with how other optional integrations (e.g., Iceberg) are handled in this project.

github-actions · 2026-05-06T21:31:16Z

+import com.amazon.deequ.analyzers.runners.AnalyzerContext
+import com.amazon.deequ.metrics.Metric
+import com.amazon.deequ.repository._
+import com.amazonaws.http.{AmazonHttpClient, DefaultErrorResponseHandler,


Several imports reference classes that don't exist in aws-java-sdk-core version 1.12.128 or use incorrect package paths:

com.amazonaws.http.AmazonHttpClient — this is an internal/private class not part of the public SDK API and may not be available depending on the SDK version.

com.amazonaws.http.DefaultErrorResponseHandler — same issue.

com.amazonaws.http.ExecutionContext — same issue.

com.google.common.collect.ImmutableList — Guava is a transitive dependency of Spark but relying on it directly is fragile.

Using internal AWS SDK classes makes this code brittle and likely to break across SDK versions. Consider using a standard HTTP client (e.g., java.net.HttpURLConnection, Apache HttpClient, or the AWS SDK's high-level AmazonApiGateway client) instead.

github-actions · 2026-05-06T21:31:16Z

+
+    writeRequest.setContent(new ByteArrayInputStream(serializedResult.getBytes("UTF-8")))
+
+    apiHelper.writeHttpRequest(writeRequest)


The writeRequest.setContent(...) call mutates the writeRequest object that was passed into the constructor. If save is called multiple times, the content from the previous call is overwritten. More critically, this only serializes the current AnalysisResult — it does not merge with previously saved results. Unlike FileSystemMetricsRepository which reads existing data and appends, this implementation will lose previously saved results for different ResultKeys unless the backing API handles accumulation. This is a semantic difference from other MetricsRepository implementations that could surprise users.

github-actions · 2026-05-06T21:31:16Z

+   * Get a AnalyzerContext saved using exactly the same resultKey if present
+   */
+  override def loadByKey(resultKey: ResultKey): Option[AnalyzerContext] = {
+    load().get().find(_.resultKey == resultKey).map(_.analyzerContext)


loadByKey calls load().get() which makes a full HTTP GET request and deserializes all results, then filters client-side. This is potentially very expensive for large metric histories. At minimum, document this behavior.

github-actions · 2026-05-06T21:31:16Z

+import java.time.{LocalDate, ZoneOffset}
+import java.util.concurrent.ConcurrentHashMap
+import scala.util.{Failure, Success}
+import scala.collection.JavaConverters._


Deprecated import scala.collection.JavaConversions._ should be replaced with scala.collection.JavaConverters._ (or scala.jdk.CollectionConverters._ for Scala 2.13+). JavaConversions provides implicit conversions that can cause subtle bugs and is deprecated.

github-actions · 2026-05-06T21:31:16Z

+  private val mapRepo = new ConcurrentHashMap[ResultKey, AnalysisResult]()
+
+  override def writeHttpRequest(writeRequest: Request[Void]): Unit = {
+    val contentString = Option(IOUtils.toString(writeRequest.getContent, "UTF-8"))


The mock writeHttpRequest stores each AnalysisResult individually by ResultKey, but readHttpRequest returns all stored results. This means the test's save method only persists a single AnalysisResult per call (the serialized JSON from AnalysisResultSerde.serialize wraps it in a Seq of one element). This matches the production code's behavior but confirms the concern: the production save method does NOT accumulate results — each save overwrites the endpoint with only the latest result. The tests pass because the mock accumulates in a ConcurrentHashMap, but the real HTTP endpoint would receive only one result per POST. This is a fundamental design issue that should be addressed or clearly documented.

github-actions · 2026-07-06T10:32:16Z

This PR has been inactive for 60 days. It will be closed in 14 days if there is no further activity. If you are still working on this, please push an update or comment to keep it open.

cmachgodaddy and others added 7 commits January 10, 2023 15:05

Adding this dockerfile to be able to build the deequ

5c43d00

adding RestMetricsRepository

ecfeccb

fixing format

1e22904

finished unit tests

5ad0ab2

clean up and adding comments

4b5a80d

fixing style

eab065d

Adding new MetricRepository for reading and writing data through REST

45991ff

Adding new MetricRepository for reading and writing data through REST

github-actions Bot requested changes May 6, 2026

View reviewed changes

github-actions Bot added the stale label Jul 6, 2026

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Adding new MetricRepository to read and write data via REST#445

Adding new MetricRepository to read and write data via REST#445
cmachgodaddy wants to merge 7 commits into
awslabs:masterfrom
gdcorp-dna:master

cmachgodaddy commented Jan 12, 2023

Uh oh!

cmachgodaddy commented Jan 12, 2023

Uh oh!

rdsharma26 commented Jan 25, 2023

Uh oh!

rdsharma26 commented Jul 5, 2023

Uh oh!

cmachgodaddy commented Jun 13, 2024

Uh oh!

github-actions Bot left a comment

Uh oh!

github-actions Bot May 6, 2026

Uh oh!

github-actions Bot May 6, 2026

Uh oh!

github-actions Bot May 6, 2026

Uh oh!

github-actions Bot May 6, 2026

Uh oh!

github-actions Bot May 6, 2026

Uh oh!

github-actions Bot May 6, 2026

Uh oh!

github-actions Bot commented Jul 6, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants


		writeRequest.setContent(new ByteArrayInputStream(serializedResult.getBytes("UTF-8")))

		apiHelper.writeHttpRequest(writeRequest)

Uh oh!

Conversation

cmachgodaddy commented Jan 12, 2023

Uh oh!

cmachgodaddy commented Jan 12, 2023

Uh oh!

rdsharma26 commented Jan 25, 2023

Uh oh!

rdsharma26 commented Jul 5, 2023

Uh oh!

cmachgodaddy commented Jun 13, 2024

Uh oh!

github-actions Bot left a comment

Choose a reason for hiding this comment

Uh oh!

github-actions Bot May 6, 2026

Choose a reason for hiding this comment

Uh oh!

github-actions Bot May 6, 2026

Choose a reason for hiding this comment

Uh oh!

github-actions Bot May 6, 2026

Choose a reason for hiding this comment

Uh oh!

github-actions Bot May 6, 2026

Choose a reason for hiding this comment

Uh oh!

github-actions Bot May 6, 2026

Choose a reason for hiding this comment

Uh oh!

github-actions Bot May 6, 2026

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Jul 6, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants