distributed FDs

stanford-futuredata · May 24, 2018 · a4fb4d9 · a4fb4d9
1 parent 9dc864d
commit a4fb4d9
Show file tree

Hide file tree

Showing 4 changed files with 34 additions and 2 deletions.
diff --git a/core/src/main/java/edu/stanford/futuredata/macrobase/pipeline/BasicBatchPipeline.java b/core/src/main/java/edu/stanford/futuredata/macrobase/pipeline/BasicBatchPipeline.java
@@ -196,6 +196,8 @@ private DistributedBatchSummarizer getDistributedSummarizer(String outlierColumn
                 summarizer.setAttributes(attributes);
                 summarizer.setMinSupport(minSupport);
                 summarizer.setMinRatioMetric(minRiskRatio);
+                summarizer.setFDUsage(useFDs);
+                summarizer.setFDValues(functionalDependencies);
                 summarizer.setNumPartitions(distributedNumPartitions);
                 return summarizer;
             }

diff --git a/...tanford/futuredata/macrobase/distributed/analysis/summary/DistributedBatchSummarizer.java b/...tanford/futuredata/macrobase/distributed/analysis/summary/DistributedBatchSummarizer.java
@@ -24,6 +24,8 @@ public abstract class DistributedBatchSummarizer implements Operator<Distributed
     protected List<String> attributes = new ArrayList<>();
     protected int numThreads = Runtime.getRuntime().availableProcessors();
     protected String ratioMetric = "global_ratio";
+    protected boolean useFDs = false;
+    protected int[] functionalDependencies;
 
     /**
      * Adjust this to tune the significance (e.g. number of rows affected) of the results returned.
@@ -64,4 +66,15 @@ public DistributedBatchSummarizer setRatioMetric(final String ratioMetric) {
         this.ratioMetric = ratioMetric;
         return this;
     }
+
+    public DistributedBatchSummarizer setFDUsage(final boolean useFDs) {
+        this.useFDs = useFDs;
+        return this;
+    }
+
+    public DistributedBatchSummarizer setFDValues(final int[] functionalDependencies) {
+        this.functionalDependencies = functionalDependencies;
+        return this;
+    }
+
 }
diff --git a/.../macrobase/distributed/analysis/summary/aplinearDistributed/APLSummarizerDistributed.java b/.../macrobase/distributed/analysis/summary/aplinearDistributed/APLSummarizerDistributed.java
@@ -169,7 +169,9 @@ private void processInternal(JavaPairRDD<String[], double[]> partitionedDataFram
                 encoder.getOutlierList(),
                 encoder.getColCardinalities(),
                 qualityMetricList,
-                thresholds
+                thresholds,
+                useFDs,
+                functionalDependencies
         );
         log.info("Number of results: {}", aplResults.size());
 

diff --git a/.../macrobase/distributed/analysis/summary/aplinearDistributed/APrioriLinearDistributed.java b/.../macrobase/distributed/analysis/summary/aplinearDistributed/APrioriLinearDistributed.java
@@ -33,7 +33,9 @@ public static List<APLExplanationResult> explain(
             ArrayList<Integer>[] outlierList,
             int[] colCardinalities,
             List<QualityMetric> argQualityMetrics,
-            List<Double> argThresholds
+            List<Double> argThresholds,
+            boolean useFDs,
+            int[] functionalDependencies
     ) {
 
         Logger log = LoggerFactory.getLogger("APLSummarizerDistributed");
@@ -189,6 +191,10 @@ public static List<APLExplanationResult> explain(
                     for (int colNumOne = 0; colNumOne < numColumns; colNumOne++) {
                         int[] curColumnOneAttributes = attributesForThread[colNumOne];
                         for (int colNumTwo = colNumOne + 1; colNumTwo < numColumns; colNumTwo++) {
+                            //if FDs are enabled, and these two attribute cols are FDs, skip
+                            if (useFDs && ((functionalDependencies[colNumOne] & (1<<colNumTwo)) == (1<<colNumTwo))) {
+                                continue;
+                            }
                             int[] curColumnTwoAttributes = attributesForThread[colNumTwo];
                             if (colCardinalities[colNumOne] < AttributeEncoder.cardinalityThreshold &&
                                     colCardinalities[colNumOne] < AttributeEncoder.cardinalityThreshold &&
@@ -210,8 +216,17 @@ public static List<APLExplanationResult> explain(
                     for (int colNumOne = 0; colNumOne < numColumns; colNumOne++) {
                         int[] curColumnOneAttributes = attributesForThread[colNumOne % numColumns];
                         for (int colNumTwo = colNumOne + 1; colNumTwo < numColumns; colNumTwo++) {
+                            //if FD on and attributes 1 and 2 are FDs, skip
+                            if (useFDs && ((functionalDependencies[colNumOne] & (1<<colNumTwo)) == (1<<colNumTwo))) {
+                                continue;
+                            }
                             int[] curColumnTwoAttributes = attributesForThread[colNumTwo % numColumns];
                             for (int colNumThree = colNumTwo + 1; colNumThree < numColumns; colNumThree++) {
+                                //if FD on and attribute 3 is FD w/ 1 or 2, skip
+                                if (useFDs && (((functionalDependencies[colNumOne] & (1 << colNumThree)) == (1 << colNumThree))
+                                        || ((functionalDependencies[colNumTwo] & (1 << colNumThree)) == (1 << colNumThree)))) {
+                                    continue;
+                                }
                                 int[] curColumnThreeAttributes = attributesForThread[colNumThree % numColumns];
                                 if (colCardinalities[colNumOne] < AttributeEncoder.cardinalityThreshold &&
                                         colCardinalities[colNumOne] < AttributeEncoder.cardinalityThreshold &&