diff --git a/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/MLContext.java b/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/MLContext.java index 534f873e5..cf799a23a 100644 --- a/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/MLContext.java +++ b/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/MLContext.java @@ -63,8 +63,8 @@ public void execute(final WayangPlan wayangPlan, final String... udfJars) { final Optional originalOption = config.getOptionalStringProperty("wayang.ml.experience.original"); final OneHotMappings mappings = new OneHotMappings(); - final TreeEncoder encoder = new TreeEncoder(mappings); - final String original = originalOption.orElse(encoder.encode(wayangPlan, wayangJob.getOptimizationContext(), false).toString()); + final TreeEncoder encoder = new TreeEncoder(mappings, config); + final String original = originalOption.orElse(encoder.encode(wayangPlan, wayangJob.getOptimizationContext()).toString()); final Optional choicesOption = config .getOptionalStringProperty("wayang.ml.experience.with-platforms"); diff --git a/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/MachineLearning.java b/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/MachineLearning.java index 80ebc1f67..f48f8fa50 100644 --- a/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/MachineLearning.java +++ b/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/MachineLearning.java @@ -18,6 +18,7 @@ package org.apache.wayang.ml; +import java.io.InputStream; import java.util.Collection; import java.util.Collections; @@ -26,6 +27,7 @@ import org.apache.wayang.core.optimizer.channels.ChannelConversion; import org.apache.wayang.core.platform.Platform; import org.apache.wayang.core.plugin.Plugin; +import org.apache.wayang.core.util.ReflectionUtils; import org.apache.wayang.java.platform.JavaPlatform; import org.apache.wayang.spark.platform.SparkPlatform; @@ -34,6 +36,8 @@ */ public class MachineLearning { + private static final String DEFAULT_CONFIG_FILE = "wayang-ml-defaults.properties"; + /** * Enables use with the {@link JavaPlatform} and {@link SparkPlatform}. */ @@ -56,6 +60,10 @@ public Collection getChannelConversions() { @Override public void setProperties(final Configuration configuration) { + final InputStream is = ReflectionUtils.loadResource(DEFAULT_CONFIG_FILE); + if (is != null) { + configuration.load(is); + } } }; diff --git a/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/encoding/TreeEncoder.java b/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/encoding/TreeEncoder.java index cff4430f2..51fa12c5c 100644 --- a/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/encoding/TreeEncoder.java +++ b/wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/encoding/TreeEncoder.java @@ -27,6 +27,7 @@ import java.util.Map; import java.util.Queue; +import org.apache.wayang.core.api.Configuration; import org.apache.wayang.core.api.exception.WayangException; import org.apache.wayang.core.optimizer.OptimizationContext; import org.apache.wayang.core.optimizer.enumeration.PlanImplementation; @@ -40,17 +41,50 @@ import org.apache.wayang.core.platform.Junction; public class TreeEncoder { + + public static final String ENCODE_IDS_PROPERTY = "wayang.ml.encoding.encode-ids"; + private final OneHotMappings mappings; + private Configuration configuration; + public TreeEncoder(final OneHotMappings mappings) { + this(mappings, null); + } + + public TreeEncoder(final OneHotMappings mappings, final Configuration configuration) { this.mappings = mappings; + this.configuration = configuration; } public OneHotMappings getMappings() { return this.mappings; } + public Configuration getConfiguration() { + return this.configuration; + } + + public void setConfiguration(final Configuration configuration) { + this.configuration = configuration; + } + + private boolean resolveEncodeIds(final OptimizationContext optimizationContext) { + final Configuration config = this.configuration != null + ? this.configuration + : (optimizationContext != null ? optimizationContext.getConfiguration() : null); + if (config != null) { + return config.getOptionalBooleanProperty(ENCODE_IDS_PROPERTY) + .orElseGet(() -> config.getBooleanProperty("wayang.ml.encode-ids", false)); + } + return false; + } + public TreeNode encode(final PlanImplementation plan) { + return this.encode(plan, this.resolveEncodeIds(plan.getOptimizationContext())); + } + + public TreeNode encode(final PlanImplementation plan, final boolean encodeIds) { final List result = new ArrayList(); final HashMap> tree = new HashMap<>(); @@ -58,9 +92,6 @@ public TreeNode encode(final PlanImplementation plan) { final Map, Junction> junctions = plan.getJunctions(); - // TODO: convert to config - final boolean encodeIds = false; - for (final Operator sink : sinks) { final TreeNode sinkNode = traversePIOperator(sink, plan.getOptimizationContext(), encodeIds, junctions, tree); @@ -77,6 +108,10 @@ public TreeNode encode(final PlanImplementation plan) { return resultNode; } + public TreeNode encode(final WayangPlan plan, final OptimizationContext optimizationContext) { + return this.encode(plan, optimizationContext, this.resolveEncodeIds(optimizationContext)); + } + public TreeNode encode(final WayangPlan plan, final OptimizationContext optimizationContext, final boolean encodeIds) { final List result = new ArrayList(); diff --git a/wayang-plugins/wayang-ml/src/main/resources/wayang-ml-defaults.properties b/wayang-plugins/wayang-ml/src/main/resources/wayang-ml-defaults.properties index 822169eb8..109d24d85 100644 --- a/wayang-plugins/wayang-ml/src/main/resources/wayang-ml-defaults.properties +++ b/wayang-plugins/wayang-ml/src/main/resources/wayang-ml-defaults.properties @@ -73,4 +73,5 @@ wayang.ml.experience.enabled = false wayang.ml.executions.file = /var/www/html/data/executions.txt wayang.ml.optimizations.file = /var/www/html/data/optmizations.txt wayang.ml.experience.file = /var/www/html/data/experience/experience-vae.txt +wayang.ml.encoding.encode-ids = false org.apache.logging.log4j.level = INFO diff --git a/wayang-plugins/wayang-ml/src/test/java/org/apache/wayang/ml/test/TreeEncoderTest.java b/wayang-plugins/wayang-ml/src/test/java/org/apache/wayang/ml/test/TreeEncoderTest.java index 31ea7de01..9add2d072 100644 --- a/wayang-plugins/wayang-ml/src/test/java/org/apache/wayang/ml/test/TreeEncoderTest.java +++ b/wayang-plugins/wayang-ml/src/test/java/org/apache/wayang/ml/test/TreeEncoderTest.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.net.URISyntaxException; +import java.util.Collection; import java.util.LinkedList; import java.util.List; @@ -28,6 +29,7 @@ import org.apache.wayang.core.api.Configuration; import org.apache.wayang.core.api.Job; import org.apache.wayang.core.api.WayangContext; +import org.apache.wayang.core.optimizer.enumeration.PlanImplementation; import org.apache.wayang.core.plan.executionplan.ExecutionPlan; import org.apache.wayang.core.plan.wayangplan.WayangPlan; import org.apache.wayang.java.Java; @@ -57,5 +59,100 @@ public void testTreeEncoding() throws IOException, URISyntaxException { Assertions.assertNotNull(exPlan); Assertions.assertNotNull(encoded); + Assertions.assertEquals(0L, encoded.encoded[0]); + } + + @Test + public void testEncodePlanImplementationWithDefaultConfig() throws IOException, URISyntaxException { + final List> collector = new LinkedList<>(); + final Configuration config = new Configuration(); + final String filePath = JavaExecutionMLTest.class.getResource("/README.md").toURI().toString(); + final WayangPlan wayangPlan = createWayangPlan(filePath, collector); + final WayangContext wayangContext = new WayangContext(config); + wayangContext.register(Java.basicPlugin()); + wayangContext.register(Spark.basicPlugin()); + + final Collection planImplementations = buildPlanImplementations(wayangPlan, wayangContext); + Assertions.assertFalse(planImplementations.isEmpty()); + + final TreeEncoder encoder = new TreeEncoder(new OneHotMappings()); + for (final PlanImplementation planImplementation : planImplementations) { + final TreeNode encoded = encoder.encode(planImplementation); + Assertions.assertNotNull(encoded); + Assertions.assertEquals(0L, encoded.encoded[0]); + } + } + + @Test + public void testEncodePlanImplementationWithEncodeIdsEnabled() throws IOException, URISyntaxException { + final List> collector = new LinkedList<>(); + final Configuration config = new Configuration(); + config.setProperty(TreeEncoder.ENCODE_IDS_PROPERTY, "true"); + final String filePath = JavaExecutionMLTest.class.getResource("/README.md").toURI().toString(); + final WayangPlan wayangPlan = createWayangPlan(filePath, collector); + final WayangContext wayangContext = new WayangContext(config); + wayangContext.register(Java.basicPlugin()); + wayangContext.register(Spark.basicPlugin()); + + final Collection planImplementations = buildPlanImplementations(wayangPlan, wayangContext); + Assertions.assertFalse(planImplementations.isEmpty()); + + final TreeEncoder encoder = new TreeEncoder(new OneHotMappings()); + for (final PlanImplementation planImplementation : planImplementations) { + final TreeNode encoded = encoder.encode(planImplementation); + Assertions.assertNotNull(encoded); + Assertions.assertNotEquals(0L, encoded.encoded[0]); + } + } + + @Test + public void testEncodeWayangPlanWithConfiguration() throws IOException, URISyntaxException { + final List> collector = new LinkedList<>(); + final Configuration config = new Configuration(); + config.setProperty("wayang.ml.encode-ids", "true"); + final String filePath = JavaExecutionMLTest.class.getResource("/README.md").toURI().toString(); + final WayangPlan wayangPlan = createWayangPlan(filePath, collector); + final WayangContext wayangContext = new WayangContext(config); + final Job wayangJob = wayangContext.createJob("", wayangPlan, ""); + wayangContext.register(Java.basicPlugin()); + wayangContext.register(Spark.basicPlugin()); + + wayangJob.buildInitialExecutionPlan(); + + final TreeEncoder encoder = new TreeEncoder(new OneHotMappings()); + final TreeNode encoded = encoder.encode(wayangPlan, wayangJob.getOptimizationContext()); + + Assertions.assertNotNull(encoded); + Assertions.assertNotEquals(0L, encoded.encoded[0]); + } + + @Test + public void testTreeEncoderConstructorWithConfiguration() throws IOException, URISyntaxException { + final List> collector = new LinkedList<>(); + final Configuration config = new Configuration(); + config.setProperty(TreeEncoder.ENCODE_IDS_PROPERTY, "true"); + final String filePath = JavaExecutionMLTest.class.getResource("/README.md").toURI().toString(); + final WayangPlan wayangPlan = createWayangPlan(filePath, collector); + final WayangContext wayangContext = new WayangContext(config); + final Job wayangJob = wayangContext.createJob("", wayangPlan, ""); + wayangContext.register(Java.basicPlugin()); + wayangContext.register(Spark.basicPlugin()); + + wayangJob.buildInitialExecutionPlan(); + + final TreeEncoder encoder = new TreeEncoder(new OneHotMappings(), config); + Assertions.assertEquals(config, encoder.getConfiguration()); + + final TreeNode encoded = encoder.encode(wayangPlan, wayangJob.getOptimizationContext()); + Assertions.assertNotNull(encoded); + Assertions.assertNotEquals(0L, encoded.encoded[0]); + } + + @Test + public void testMachineLearningPluginSetProperties() { + final Configuration config = new Configuration(); + org.apache.wayang.ml.MachineLearning.plugin().setProperties(config); + Assertions.assertEquals(false, config.getBooleanProperty(TreeEncoder.ENCODE_IDS_PROPERTY)); + Assertions.assertEquals(100L, config.getLongProperty("wayang.ml.tuple.average-size")); } }