[FSTORE-1412] Feature monitoring improvements (#980)

Author: javierdlrm

java/hsfs/src/main/java/com/logicalclocks/hsfs/StatisticsConfig.java

@@ -50,6 +50,14 @@ public class StatisticsConfig {
   @Setter
   private List<String> columns = new ArrayList<>();
 
+  @Getter
+  @Setter
+  private Boolean kll = false;
+
+  @Getter
+  @Setter
+  private Integer histogramBins = null;
+
   public StatisticsConfig(Boolean enabled, Boolean histograms, Boolean correlations,
                           Boolean exactUniqueness) {
     this.enabled = enabled;

java/hsfs/src/main/java/com/logicalclocks/hsfs/metadata/FeatureDescriptiveStatistics.java

@@ -24,7 +24,7 @@
 import lombok.Setter;
 import org.json.JSONObject;
 
-import java.util.Map;
+import java.util.List;
 
 @JsonIgnoreProperties(ignoreUnknown = true)
 @NoArgsConstructor
@@ -48,8 +48,8 @@ public class FeatureDescriptiveStatistics extends RestDto<FeatureDescriptiveStat
   private Double max;
   private Double sum;
   private Double mean;
-  private Double stddev;
-  private Map<String, Double> percentiles;
+  private Double stdDev;
+  private List<Double> percentiles;
 
   // with exact uniqueness
   private Double distinctness;
@@ -122,7 +122,7 @@ public static FeatureDescriptiveStatistics fromDeequStatisticsJson(JSONObject st
       fds.setMean(statsJson.getDouble("mean"));
     }
     if (statsJson.has("stdDev")) {
-      fds.setStddev(statsJson.getDouble("stdDev"));
+      fds.setStdDev(statsJson.getDouble("stdDev"));
     }
 
     JSONObject extendedStatistics = new JSONObject();

java/pom.xml

@@ -27,7 +27,7 @@
         <lombok.version>1.18.36</lombok.version>
         <lombok.plugin.version>1.18.20.0</lombok.plugin.version>
         <fasterxml.jackson.databind.version>2.10.0</fasterxml.jackson.databind.version>
-        <deequ.version>1.1.0.6-SNAPSHOT</deequ.version>
+        <datasketches.version>6.2.0</datasketches.version>
         <spark.version>3.5.5.1</spark.version>
         <hudi.version>1.0.2.1</hudi.version>
         <awssdk.version>2.10.40</awssdk.version>
@@ -295,7 +295,6 @@
         <profile>
             <id>spark-3.3</id>
             <properties>
-                <deequ.version>2.0.4.0-spark-3.3</deequ.version>
                 <artifact.spark.version>spark3.3</artifact.spark.version>
             </properties>
         </profile>

java/spark/pom.xml

@@ -52,20 +52,12 @@
             </exclusions>
         </dependency>
 
+        <!-- datasketches-java: KllDoublesSketch for mergeable percentile sketches.
+             Replaces the Deequ-based profiling path as of Phase 1.5 (FSTORE-1412). -->
         <dependency>
-            <groupId>com.logicalclocks</groupId>
-            <artifactId>deequ_${scala-short.version}</artifactId>
-            <version>${deequ.version}</version>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.apache.spark</groupId>
-                    <artifactId>*</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>org.scala-lang</groupId>
-                    <artifactId>*</artifactId>
-                </exclusion>
-            </exclusions>
+            <groupId>org.apache.datasketches</groupId>
+            <artifactId>datasketches-java</artifactId>
+            <version>${datasketches.version}</version>
         </dependency>
 
         <dependency>

java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java

@@ -84,9 +84,6 @@
 import org.apache.spark.sql.types.TimestampType;
 import org.json.JSONObject;
 
-import com.amazon.deequ.profiles.ColumnProfilerRunBuilder;
-import com.amazon.deequ.profiles.ColumnProfilerRunner;
-import com.amazon.deequ.profiles.ColumnProfiles;
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
@@ -115,6 +112,7 @@
 import com.logicalclocks.hsfs.spark.TrainingDataset;
 import com.logicalclocks.hsfs.spark.constructor.Query;
 import com.logicalclocks.hsfs.spark.engine.hudi.HudiEngine;
+import com.logicalclocks.hsfs.spark.engine.profile.ColumnProfiler;
 import com.logicalclocks.hsfs.spark.util.StorageConnectorUtils;
 import com.logicalclocks.hsfs.util.Constants;
 
@@ -666,28 +664,22 @@ private void writeSparkDataset(FeatureGroupBase featureGroup, Dataset<Row> datas
         .saveAsTable(featureGroupUtils.getTableName(featureGroup));
   }
 
+  public String profile(Dataset<Row> df, List<String> restrictToColumns, Boolean correlation,
+      Boolean histogram, Boolean exactUniqueness, Boolean kll, Integer histogramBins) {
+    // defaults aligned with the prior Deequ-backed implementation; preserved for training-dataset
+    // callers where the backend doesn't set them.
+    boolean correlationFlag = correlation == null ? true : correlation;
+    boolean histogramFlag = histogram == null ? true : histogram;
+    boolean exactUniquenessFlag = exactUniqueness == null ? true : exactUniqueness;
+    boolean kllFlag = kll != null && kll;
+    int binCount = histogramBins != null ? histogramBins : 20;
+    return new ColumnProfiler().profile(df, restrictToColumns, correlationFlag, histogramFlag,
+        binCount, exactUniquenessFlag, kllFlag);
+  }
+
   public String profile(Dataset<Row> df, List<String> restrictToColumns, Boolean correlation,
       Boolean histogram, Boolean exactUniqueness) {
-    // only needed for training datasets, as the backend is not setting the defaults
-    if (correlation == null) {
-      correlation = true;
-    }
-    if (histogram == null) {
-      histogram = true;
-    }
-    if (exactUniqueness == null) {
-      exactUniqueness = true;
-    }
-    ColumnProfilerRunBuilder runner = new ColumnProfilerRunner()
-                                            .onData(df)
-                                            .withCorrelation(correlation, 100)
-                                            .withHistogram(histogram, 20)
-                                            .withExactUniqueness(exactUniqueness);
-    if (restrictToColumns != null && !restrictToColumns.isEmpty()) {
-      runner.restrictToColumns(JavaConverters.asScalaIteratorConverter(restrictToColumns.iterator()).asScala().toSeq());
-    }
-    ColumnProfiles result = runner.run();
-    return ColumnProfiles.toJson(result.profiles().values().toSeq(), result.numRecords());
+    return profile(df, restrictToColumns, correlation, histogram, exactUniqueness, null, null);
   }
 
   public String profile(Dataset<Row> df, List<String> restrictToColumns, Boolean correlation, Boolean histogram) {