@Test public void testPlanGenerator() throws InjectionException, IOException, URISyntaxException, ClassNotFoundException { final MISTQueryBuilder queryBuilder = new MISTQueryBuilder(); queryBuilder.setApplicationId(TestParameters.SUPER_GROUP_ID); queryBuilder.socketTextStream(TestParameters.LOCAL_TEXT_SOCKET_SOURCE_CONF) .flatMap(s -> Arrays.asList(s.split(" "))) .filter(s -> s.startsWith("A")) .map(s -> new Tuple2<>(s, 1)) .reduceByKey(0, String.class, (Integer x, Integer y) -> x + y) .textSocketOutput(TestParameters.HOST, TestParameters.SINK_PORT); final MISTQuery query = queryBuilder.build(); final Tuple<List<AvroVertex>, List<Edge>> serializedDag = query.getAvroOperatorDag(); final AvroDag.Builder avroDagBuilder = AvroDag.newBuilder(); final AvroDag avroChainedDag = avroDagBuilder .setAppId(TestParameters.SUPER_GROUP_ID) .setQueryId(TestParameters.QUERY_ID) .setJarPaths(new ArrayList<>()) .setAvroVertices(serializedDag.getKey()) .setEdges(serializedDag.getValue()) .build(); final JavaConfigurationBuilder jcb = Tang.Factory.getTang().newConfigurationBuilder(); jcb.bindNamedParameter(TaskHostname.class, "127.0.0.1"); final Injector injector = Tang.Factory.getTang().newInjector(jcb.build()); final ConfigDagGenerator configDagGenerator = injector.getInstance(ConfigDagGenerator.class); final DagGenerator dagGenerator = injector.getInstance(DagGenerator.class); final Tuple<String, AvroDag> tuple = new Tuple<>("query-test", avroChainedDag); final DAG<ConfigVertex, MISTEdge> configDag = configDagGenerator.generate(tuple.getValue()); final ExecutionDag executionDag = dagGenerator.generate(configDag, new LinkedList<>()); final DAG<ExecutionVertex, MISTEdge> dag = executionDag.getDag(); final Set<ExecutionVertex> sources = dag.getRootVertices(); Assert.assertEquals(1, sources.size()); Assert.assertTrue(sources.iterator().next() instanceof PhysicalSource); final PhysicalSource source = (PhysicalSource)sources.iterator().next(); final Map<ExecutionVertex, MISTEdge> nextOps = dag.getEdges(source); Assert.assertEquals(1, nextOps.size()); final PhysicalOperator flatMapOp = (PhysicalOperator)nextOps.entrySet().iterator().next().getKey(); final PhysicalOperator filterOp = (PhysicalOperator)dag.getEdges(flatMapOp).entrySet().iterator().next().getKey(); final PhysicalOperator mapOp = (PhysicalOperator)dag.getEdges(filterOp).entrySet().iterator().next().getKey(); final PhysicalOperator reduceByKeyOp = (PhysicalOperator)dag.getEdges(mapOp) .entrySet().iterator().next().getKey(); final PhysicalSink sink = (PhysicalSink)dag.getEdges(reduceByKeyOp) .entrySet().iterator().next().getKey(); Assert.assertTrue(flatMapOp.getOperator() instanceof FlatMapOperator); Assert.assertTrue(filterOp.getOperator() instanceof FilterOperator); Assert.assertTrue(mapOp.getOperator() instanceof MapOperator); Assert.assertTrue(reduceByKeyOp.getOperator() instanceof ReduceByKeyOperator); Assert.assertTrue(sink.getSink() instanceof NettyTextSink); }
@Override public ExecutionDag generate(final DAG<ConfigVertex, MISTEdge> configDag, final List<String> jarFilePaths) throws IOException, ClassNotFoundException { final ExecutionDag executionDag = new ExecutionDag(new AdjacentListConcurrentMapDAG<>()); final URL[] urls = SerializeUtils.getJarFileURLs(jarFilePaths); final ClassLoader classLoader = classLoaderProvider.newInstance(urls); final Map<ConfigVertex, ExecutionVertex> created = new HashMap<>(configDag.numberOfVertices()); for (final ConfigVertex source : configDag.getRootVertices()) { final ExecutionVertex currExecutionVertex = executionVertexGenerator.generate(source, urls, classLoader); executionDag.getDag().addVertex(currExecutionVertex); created.put(source, currExecutionVertex); for (final Map.Entry<ConfigVertex, MISTEdge> edges : configDag.getEdges(source).entrySet()) { final ConfigVertex childVertex = edges.getKey(); final MISTEdge edge = edges.getValue(); dfsCreation(currExecutionVertex, edge, childVertex, created, configDag, executionDag, urls, classLoader); } } return executionDag; }
DefaultDagGeneratorImpl implements DagGenerator { @Override public ExecutionDag generate(final DAG<ConfigVertex, MISTEdge> configDag, final List<String> jarFilePaths) throws IOException, ClassNotFoundException { final ExecutionDag executionDag = new ExecutionDag(new AdjacentListConcurrentMapDAG<>()); final URL[] urls = SerializeUtils.getJarFileURLs(jarFilePaths); final ClassLoader classLoader = classLoaderProvider.newInstance(urls); final Map<ConfigVertex, ExecutionVertex> created = new HashMap<>(configDag.numberOfVertices()); for (final ConfigVertex source : configDag.getRootVertices()) { final ExecutionVertex currExecutionVertex = executionVertexGenerator.generate(source, urls, classLoader); executionDag.getDag().addVertex(currExecutionVertex); created.put(source, currExecutionVertex); for (final Map.Entry<ConfigVertex, MISTEdge> edges : configDag.getEdges(source).entrySet()) { final ConfigVertex childVertex = edges.getKey(); final MISTEdge edge = edges.getValue(); dfsCreation(currExecutionVertex, edge, childVertex, created, configDag, executionDag, urls, classLoader); } } return executionDag; } }
DefaultDagGeneratorImpl implements DagGenerator { @Override public ExecutionDag generate(final DAG<ConfigVertex, MISTEdge> configDag, final List<String> jarFilePaths) throws IOException, ClassNotFoundException { final ExecutionDag executionDag = new ExecutionDag(new AdjacentListConcurrentMapDAG<>()); final URL[] urls = SerializeUtils.getJarFileURLs(jarFilePaths); final ClassLoader classLoader = classLoaderProvider.newInstance(urls); final Map<ConfigVertex, ExecutionVertex> created = new HashMap<>(configDag.numberOfVertices()); for (final ConfigVertex source : configDag.getRootVertices()) { final ExecutionVertex currExecutionVertex = executionVertexGenerator.generate(source, urls, classLoader); executionDag.getDag().addVertex(currExecutionVertex); created.put(source, currExecutionVertex); for (final Map.Entry<ConfigVertex, MISTEdge> edges : configDag.getEdges(source).entrySet()) { final ConfigVertex childVertex = edges.getKey(); final MISTEdge edge = edges.getValue(); dfsCreation(currExecutionVertex, edge, childVertex, created, configDag, executionDag, urls, classLoader); } } return executionDag; } @Inject private  DefaultDagGeneratorImpl(final ClassLoaderProvider classLoaderProvider,
                                  final ExecutionVertexGenerator executionVertexGenerator); }
DefaultDagGeneratorImpl implements DagGenerator { @Override public ExecutionDag generate(final DAG<ConfigVertex, MISTEdge> configDag, final List<String> jarFilePaths) throws IOException, ClassNotFoundException { final ExecutionDag executionDag = new ExecutionDag(new AdjacentListConcurrentMapDAG<>()); final URL[] urls = SerializeUtils.getJarFileURLs(jarFilePaths); final ClassLoader classLoader = classLoaderProvider.newInstance(urls); final Map<ConfigVertex, ExecutionVertex> created = new HashMap<>(configDag.numberOfVertices()); for (final ConfigVertex source : configDag.getRootVertices()) { final ExecutionVertex currExecutionVertex = executionVertexGenerator.generate(source, urls, classLoader); executionDag.getDag().addVertex(currExecutionVertex); created.put(source, currExecutionVertex); for (final Map.Entry<ConfigVertex, MISTEdge> edges : configDag.getEdges(source).entrySet()) { final ConfigVertex childVertex = edges.getKey(); final MISTEdge edge = edges.getValue(); dfsCreation(currExecutionVertex, edge, childVertex, created, configDag, executionDag, urls, classLoader); } } return executionDag; } @Inject private  DefaultDagGeneratorImpl(final ClassLoaderProvider classLoaderProvider,
                                  final ExecutionVertexGenerator executionVertexGenerator); @Override ExecutionDag generate(final DAG<ConfigVertex, MISTEdge> configDag,
                               final List<String> jarFilePaths); }
DefaultDagGeneratorImpl implements DagGenerator { @Override public ExecutionDag generate(final DAG<ConfigVertex, MISTEdge> configDag, final List<String> jarFilePaths) throws IOException, ClassNotFoundException { final ExecutionDag executionDag = new ExecutionDag(new AdjacentListConcurrentMapDAG<>()); final URL[] urls = SerializeUtils.getJarFileURLs(jarFilePaths); final ClassLoader classLoader = classLoaderProvider.newInstance(urls); final Map<ConfigVertex, ExecutionVertex> created = new HashMap<>(configDag.numberOfVertices()); for (final ConfigVertex source : configDag.getRootVertices()) { final ExecutionVertex currExecutionVertex = executionVertexGenerator.generate(source, urls, classLoader); executionDag.getDag().addVertex(currExecutionVertex); created.put(source, currExecutionVertex); for (final Map.Entry<ConfigVertex, MISTEdge> edges : configDag.getEdges(source).entrySet()) { final ConfigVertex childVertex = edges.getKey(); final MISTEdge edge = edges.getValue(); dfsCreation(currExecutionVertex, edge, childVertex, created, configDag, executionDag, urls, classLoader); } } return executionDag; } @Inject private  DefaultDagGeneratorImpl(final ClassLoaderProvider classLoaderProvider,
                                  final ExecutionVertexGenerator executionVertexGenerator); @Override ExecutionDag generate(final DAG<ConfigVertex, MISTEdge> configDag,
                               final List<String> jarFilePaths);  }