diff --git a/.travis.yml b/.travis.yml
index efb62ac..d00414c 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -48,6 +48,7 @@
   - echo '  </server>' >> ~/.m2/settings.xml
   - echo ' </servers>' >> ~/.m2/settings.xml
   - echo '</settings>' >> ~/.m2/settings.xml
+  - sudo apt-get install -y protobuf-compiler
   - mkdir -p travis/tmp
   - echo "#" >> travis/wayang.properties
   - echo "# Licensed to the Apache Software Foundation (ASF) under one or more" >> travis/wayang.properties
diff --git a/pom.xml b/pom.xml
index 3ff48f9..2f7fadd 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1248,6 +1248,8 @@
                         <exclude>**/README.md</exclude>
                         <exclude>**/general-todos.md</exclude>
                         <exclude>**/scala_1*</exclude>
+
+                        <exclude>**/*pb2.py</exclude>
                     </excludes>
                 </configuration>
             </plugin>
@@ -1278,7 +1280,7 @@
                     </includedLicenses>
                     <failOnBlacklist>true</failOnBlacklist>
                     <excludedGroups>
-                        org.apache.spark.*|org.apache.hadoop.*|org.apache.giraph.*|org.antlr.*|junit.*|org.graphchi.*
+                        org.apache.spark.*|org.apache.hadoop.*|org.apache.giraph.*|org.antlr.*|junit.*|org.graphchi.*|org.springframework.*
                     </excludedGroups>
                     <excludeTransitiveDependencies>true</excludeTransitiveDependencies>
                 </configuration>
diff --git a/pywayang/config/__init__.py b/pywayang/config/__init__.py
new file mode 100644
index 0000000..008475c
--- /dev/null
+++ b/pywayang/config/__init__.py
@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from config.config_reader import get_source_types
+from config.config_reader import get_sink_types
+from config.config_reader import get_boundary_types
diff --git a/pywayang/config/config_reader.py b/pywayang/config/config_reader.py
new file mode 100644
index 0000000..c8f5873
--- /dev/null
+++ b/pywayang/config/config_reader.py
@@ -0,0 +1,51 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import configparser
+import os
+
+
+def get_boundary_types():
+    config = configparser.ConfigParser()
+    config.sections()
+    config.read('../config/pywayang_config.ini')
+    boundary_types = dict(config.items('BOUNDARY_TYPES'))
+    boundary_types.pop("variable_to_access")
+    return boundary_types.values()
+
+
+def get_source_types():
+    config = configparser.ConfigParser()
+    #print("path: ", os.getcwd())
+    config.read("../config/pywayang_config.ini")
+    source_types = dict(config.items('SOURCE_TYPES'))
+    source_types.pop("variable_to_access")
+    return source_types.values()
+    #sections_list = config.sections()
+    #for section in sections_list:
+    #    print(section)
+    #print("source_types")
+    #for x in source_types.values():
+    #    print(x)
+
+def get_sink_types():
+    config = configparser.ConfigParser()
+    #print("path: ", os.getcwd())
+    config.read("../config/pywayang_config.ini")
+    sink_types = dict(config.items('SINK_TYPES'))
+    sink_types.pop("variable_to_access")
+    return sink_types.values()
\ No newline at end of file
diff --git a/pywayang/config/pywayang_config.ini b/pywayang/config/pywayang_config.ini
new file mode 100644
index 0000000..78cc2b4
--- /dev/null
+++ b/pywayang/config/pywayang_config.ini
@@ -0,0 +1,38 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+[DEFAULT]
+variable_to_access = value
+
+[INPUT]
+txnname_mod = string1
+txnmemo_mod = string2
+
+[MODIFY]
+txnname_mod = string3
+txnmemo_mod = string4
+
+[BOUNDARY_TYPES]
+boundary_type_1 = union
+
+[SOURCE_TYPES]
+source_type_1 = source
+source_type_2 = text
+
+[SINK_TYPES]
+sink_type_1 = sink
+sink_type_2 = sonk
\ No newline at end of file
diff --git a/pywayang/graph/__init__.py b/pywayang/graph/__init__.py
new file mode 100644
index 0000000..17e2deb
--- /dev/null
+++ b/pywayang/graph/__init__.py
@@ -0,0 +1,19 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import graph.graph
+import graph.node
\ No newline at end of file
diff --git a/pywayang/graph/graph.py b/pywayang/graph/graph.py
new file mode 100644
index 0000000..be7a32f
--- /dev/null
+++ b/pywayang/graph/graph.py
@@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from graph.node import Node
+import logging
+
+
+# Adjacency Matrix used to analise the plan
+class Graph:
+    def __init__(self):
+        self.graph = {}
+        self.nodes_no = 0
+        self.nodes = []
+
+    # Fills the Graph
+    def populate(self, sinks):
+        for sink in iter(sinks):
+            self.process_operator(sink)
+
+    # Add current operator and set dependencies
+    def process_operator(self, operator):
+        self.add_node(operator.operator_type, operator.id, operator)
+
+        if len(operator.previous) > 0:
+            for parent in operator.previous:
+                if parent:
+                    self.add_node(parent.operator_type, parent.id, parent)
+                    self.add_link(operator.id, parent.id, 1)
+                    self.process_operator(parent)
+
+    def add_node(self, name, id, operator):
+        if id in self.nodes:
+            return
+
+        self.nodes_no += 1
+        self.nodes.append(id)
+        new_node = Node(name, id, operator)
+
+        self.graph[id] = new_node
+
+    def add_link(self, id_child, id_parent, e):
+        if id_child in self.nodes:
+            if id_parent in self.nodes:
+                self.graph[id_child].add_predecessor(id_parent, e)
+                self.graph[id_parent].add_successor(id_child, e)
+
+    def print_adjlist(self):
+
+        for key in self.graph:
+            logging.debug("Node: ", self.graph[key].operator_type, " - ", key)
+            for key2 in self.graph[key].predecessors:
+                logging.debug("- Parent: ", self.graph[key2].operator_type, " - ", self.graph[key].predecessors[key2], " - ", key2)
+            for key2 in self.graph[key].successors:
+                logging.debug("- Child: ", self.graph[key2].operator_type, " - ", self.graph[key].successors[key2], " - ", key2)
+
+    def get_node(self, id):
+        return self.graph[id]
diff --git a/pywayang/graph/node.py b/pywayang/graph/node.py
new file mode 100644
index 0000000..d0d696f
--- /dev/null
+++ b/pywayang/graph/node.py
@@ -0,0 +1,48 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import abc
+
+
+class Element(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def accept(self, visitor, udf, orientation, last_iter):
+        pass
+
+
+# Describes an Operator in the Graph
+class Node(Element):
+    def __init__(self, operator_type, id, operator):
+        self.operator_type = operator_type
+        self.id = id
+        self.predecessors = {}
+        self.successors = {}
+        self.python_exec = operator.python_exec
+
+        # Temporal
+        self.operator = operator
+
+    def add_predecessor(self, id_parent, e):
+        self.predecessors[id_parent] = e
+
+    def add_successor(self, id_child, e):
+        self.successors[id_child] = e
+
+    # Nodes are visited by objects of class Visitant.
+    # Visitants are being used to execute a UDF through the Graph
+    def accept(self, visitor, udf, orientation, last_iter):
+        visitor.visit_node(self, udf, orientation, last_iter)
diff --git a/pywayang/graph/traversal.py b/pywayang/graph/traversal.py
new file mode 100644
index 0000000..e2dd851
--- /dev/null
+++ b/pywayang/graph/traversal.py
@@ -0,0 +1,51 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from graph.visitant import Visitant
+import logging
+
+
+# Defines how a UDF will be applied over the Graph
+class Traversal:
+
+    def __init__(self, graph, origin, udf):
+        self.graph = graph
+        self.origin = origin
+        self.udf = udf
+        self.app = Visitant(graph, [])
+
+        # Starting from Sinks or Sources sets an specific orientation
+        if origin[0].source:
+            self.orientation = "successors"
+        elif origin[0].sink:
+            self.orientation = "predecessors"
+        else:
+            logging.error("Origin point to traverse the plan wrongly defined")
+            return
+
+        for operator in iter(origin):
+            logging.debug("operator origin: " + str(operator.id))
+            node = graph.get_node(operator.id)
+            self.app.visit_node(
+                node=node,
+                udf=self.udf,
+                orientation=self.orientation,
+                last_iter=None
+            )
+
+    def get_collected_data(self):
+        return self.app.get_collection()
diff --git a/pywayang/graph/visitant.py b/pywayang/graph/visitant.py
new file mode 100644
index 0000000..3d2f874
--- /dev/null
+++ b/pywayang/graph/visitant.py
@@ -0,0 +1,52 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import abc
+import logging
+
+
+class Visitor(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def visit_node(self, node, udf, orientation, last_iter):
+        pass
+
+
+# Applies a UDF in current Node
+class Visitant(Visitor):
+
+    def __init__(self, graph, results):
+        self.collection = results
+        self.graph = graph
+
+    # UDF can store results in ApplyFunction.collection whenever its requires.
+    # last_iter has the generated current value obtained in the previous iteration
+    def visit_node(self, node, udf, orientation, last_iter):
+        logging.debug("Applying UDf" + str(orientation))
+        current_value = udf(node, last_iter, self.collection)
+        logging.debug("orientation result " + str(getattr(node, orientation)))
+        next_iter = getattr(node, orientation)
+        if len(next_iter) > 0:
+            for next_iter_id in next_iter:
+                if next_iter_id:
+                    logging.debug("next_id: " + str(next_iter_id))
+                    next_iter_node = self.graph.get_node(next_iter_id)
+                    logging.debug("next_iter_node: " + next_iter_node.operator_type + " " + str(next_iter_node.id))
+                    next_iter_node.accept(visitor=self, udf=udf, orientation=orientation, last_iter=current_value)
+        pass
+
+    def get_collection(self):
+        return self.collection
diff --git a/pywayang/orchestrator/__init__.py b/pywayang/orchestrator/__init__.py
new file mode 100644
index 0000000..ed7d0ac
--- /dev/null
+++ b/pywayang/orchestrator/__init__.py
@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import orchestrator.plan
+import orchestrator.dataquanta
+import graph.graph
diff --git a/pywayang/orchestrator/dataquanta.py b/pywayang/orchestrator/dataquanta.py
new file mode 100644
index 0000000..7d700eb
--- /dev/null
+++ b/pywayang/orchestrator/dataquanta.py
@@ -0,0 +1,330 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from orchestrator.operator import Operator
+from graph.graph import Graph
+from graph.traversal import Traversal
+from protobuf.planwriter import MessageWriter
+import itertools
+import collections
+import logging
+from functools import reduce
+import operator
+
+
+# Wraps a Source operation to create an iterable
+class DataQuantaBuilder:
+    def __init__(self, descriptor):
+        self.descriptor = descriptor
+
+    def source(self, source):
+
+        if type(source) is str:
+            source_ori = open(source, "r")
+        else:
+            source_ori = source
+        return DataQuanta(
+            Operator(
+                operator_type="source",
+                udf=source,
+                iterator=iter(source_ori),
+                previous=[],
+                python_exec=False
+            ),
+            descriptor=self.descriptor
+        )
+
+
+# Wraps an operation over an iterable
+class DataQuanta:
+    def __init__(self, operator=None, descriptor=None):
+        self.operator = operator
+        self.descriptor = descriptor
+        if self.operator.is_source():
+            self.descriptor.add_source(self.operator)
+        if self.operator.is_sink():
+            self.descriptor.add_sink(self.operator)
+
+    # Operational Functions
+    def filter(self, udf):
+        def func(iterator):
+            return filter(udf, iterator)
+
+        return DataQuanta(
+            Operator(
+                operator_type="filter",
+                udf=func,
+                previous=[self.operator],
+                python_exec=True
+            ),
+            descriptor=self.descriptor
+        )
+
+    def flatmap(self, udf):
+
+        def auxfunc(iterator):
+            return itertools.chain.from_iterable(map(udf, iterator))
+
+        def func(iterator):
+            mapped = map(udf, iterator)
+            flattened = flatten_single_dim(mapped)
+            yield from flattened
+
+        def flatten_single_dim(mapped):
+            for item in mapped:
+                for subitem in item:
+                    yield subitem
+
+        return DataQuanta(
+            Operator(
+                operator_type="flatmap",
+                udf=func,
+                previous=[self.operator],
+                python_exec=True
+            ),
+            descriptor=self.descriptor
+        )
+
+    def group_by(self, udf):
+        def func(iterator):
+            # TODO key should be given by "udf"
+            return itertools.groupby(iterator, key=operator.itemgetter(0))
+            #return itertools.groupby(sorted(iterator), key=itertools.itemgetter(0))
+
+        return DataQuanta(
+            Operator(
+                operator_type="group_by",
+                udf=func,
+                previous=[self.operator],
+                python_exec=True
+            ),
+            descriptor=self.descriptor
+        )
+
+    def map(self, udf):
+        def func(iterator):
+            return map(udf, iterator)
+
+        return DataQuanta(
+            Operator(
+                operator_type="map",
+                udf=func,
+                previous=[self.operator],
+                python_exec=True
+            ),
+            descriptor=self.descriptor
+        )
+
+    # Key specifies pivot dimensions
+    # UDF specifies reducer function
+    def reduce_by_key(self, keys, udf):
+
+        op = Operator(
+            operator_type="reduce_by_key",
+            udf=udf,
+            previous=[self.operator],
+            python_exec=False
+        )
+
+        print(len(keys), keys)
+        for i in range(0, len(keys)):
+            """if keys[i] is int:
+                op.set_parameter("vector_position|"+str(i), keys[i])
+            else:
+                op.set_parameter("dimension_key|"+str(i), keys[i])"""
+
+            # TODO maybe would be better just leave the number as key
+            op.set_parameter("dimension|"+str(i+1), keys[i])
+
+        return DataQuanta(
+            op,
+            descriptor=self.descriptor
+        )
+
+    def reduce(self, udf):
+        def func(iterator):
+            return reduce(udf, iterator)
+
+        return DataQuanta(
+            Operator(
+                operator_type="reduce",
+                udf=func,
+                previous=[self.operator],
+                python_exec=True
+            ),
+            descriptor=self.descriptor
+        )
+
+    def sink(self, path, end="\n"):
+        def consume(iterator):
+            with open(path, 'w') as f:
+                for x in iterator:
+                    f.write(str(x) + end)
+
+        def func(iterator):
+            consume(iterator)
+            # return self.__run(consume)
+
+        return DataQuanta(
+            Operator(
+                operator_type="sink",
+
+                udf=path,
+                # To execute directly uncomment
+                # udf=func,
+
+                previous=[self.operator],
+                python_exec=False
+            ),
+            descriptor=self.descriptor
+        )
+
+    def sort(self, udf):
+
+        def func(iterator):
+            return sorted(iterator, key=udf)
+
+        return DataQuanta(
+            Operator(
+                operator_type="sort",
+                udf=func,
+                previous=[self.operator],
+                python_exec=True
+            ),
+            descriptor=self.descriptor
+        )
+
+    # This function allow the union to be performed by Python
+    # Nevertheless, current configuration runs it over Java
+    def union(self, other):
+
+        def func(iterator):
+            return itertools.chain(iterator, other.operator.getIterator())
+
+        return DataQuanta(
+            Operator(
+                operator_type="union",
+                udf=func,
+                previous=[self.operator, other.operator],
+                python_exec=False
+            ),
+            descriptor=self.descriptor
+        )
+
+    def __run(self, consumer):
+        consumer(self.operator.getIterator())
+
+    # Execution Functions
+    def console(self, end="\n"):
+        def consume(iterator):
+            for x in iterator:
+                print(x, end=end)
+
+        self.__run(consume)
+
+    # Only for debugging purposes!
+    # To execute the plan directly in the program driver
+    def execute(self):
+        logging.warn("DEBUG Execution")
+        logging.info("Reminder to swap SINK UDF value from path to func")
+        logging.debug(self.operator.previous[0].operator_type)
+        if self.operator.is_sink():
+            logging.debug(self.operator.operator_type)
+            logging.debug(self.operator.udf)
+            logging.debug(len(self.operator.previous))
+            self.operator.udf(self.operator.previous[0].getIterator())
+        else:
+            logging.error("Plan must call execute from SINK type of operator")
+            raise RuntimeError
+
+    # Converts Python Functional Plan to valid Wayang Plan
+    def to_wayang_plan(self):
+
+        sinks = self.descriptor.get_sinks()
+        if len(sinks) == 0:
+            return
+
+        graph = Graph()
+        graph.populate(self.descriptor.get_sinks())
+
+        # Uncomment to check the Graph built
+        # graph.print_adjlist()
+
+        # Function to be consumed by Traverse
+        # Separates Python Plan into a List of Pipelines
+        def define_pipelines(node1, current_pipeline, collection):
+            def store_unique(pipe_to_insert):
+                for pipe in collection:
+                    if equivalent_lists(pipe, pipe_to_insert):
+                        return
+                collection.append(pipe_to_insert)
+
+            def equivalent_lists(l1, l2):
+                if collections.Counter(l1) == collections.Counter(l2):
+                    return True
+                else:
+                    return False
+
+            if not current_pipeline:
+                current_pipeline = [node1]
+
+            elif node1.operator.is_boundary():
+                store_unique(current_pipeline.copy())
+                current_pipeline.clear()
+                current_pipeline.append(node1)
+
+            else:
+                current_pipeline.append(node1)
+
+            if node1.operator.sink:
+                store_unique(current_pipeline.copy())
+                current_pipeline.clear()
+
+            return current_pipeline
+
+        # Works over the graph
+        trans = Traversal(
+            graph=graph,
+            origin=self.descriptor.get_sources(),
+            # udf=lambda x, y, z: d(x, y, z)
+            # UDF always will receive:
+            # x: a Node object,
+            # y: an object representing the result of the last iteration,
+            # z: a collection to store final results inside your UDF
+            udf=lambda x, y, z: define_pipelines(x, y, z)
+        )
+
+        # Gets the results of the traverse process
+        collected_stages = trans.get_collected_data()
+
+        # Passing the Stages to a Wayang message writer
+        writer = MessageWriter()
+        a = 0
+        # Stage is composed of class Node objects
+        for stage in collected_stages:
+            a += 1
+            logging.info("///")
+            logging.info("stage" + str(a))
+            writer.process_pipeline(stage)
+
+        writer.set_dependencies()
+
+        # Uses a file to provide the plan
+        # writer.write_message(self.descriptor)
+
+        # Send the plan to Wayang REST api directly
+        writer.send_message(self.descriptor)
diff --git a/pywayang/orchestrator/execdirectly.py b/pywayang/orchestrator/execdirectly.py
new file mode 100644
index 0000000..452ccab
--- /dev/null
+++ b/pywayang/orchestrator/execdirectly.py
@@ -0,0 +1,162 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from orchestrator.plan import Descriptor
+from orchestrator.dataquanta import DataQuantaBuilder
+import datetime
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_sort(descriptor):
+    plan = DataQuantaBuilder(descriptor)
+    sink_dataquanta = \
+        plan.source("../test/words.txt") \
+            .sort(lambda elem: elem.lower()) \
+            .sink("../test/output.txt", end="")
+    return sink_dataquanta
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_sort_filter(descriptor):
+    plan = DataQuantaBuilder(descriptor)
+    sink_dataquanta = \
+        plan.source("../test/words.txt") \
+            .sort(lambda elem: elem.lower()) \
+            .filter(lambda elem: str(elem).startswith("f")) \
+            .sink("../test/output.txt", end="")
+    return sink_dataquanta
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_filter_text(descriptor):
+    plan = DataQuantaBuilder(descriptor)
+
+    sink_dataquanta = \
+        plan.source("../test/words.txt") \
+            .filter(lambda elem: str(elem).startswith("f")) \
+            .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_filter(descriptor):
+    plan = DataQuantaBuilder(descriptor)
+
+    sink_dataquanta = \
+        plan.source("../test/numbers.txt") \
+            .filter(lambda elem: int(elem) % 2 != 0) \
+            .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_basic(descriptor):
+    plan = DataQuantaBuilder(descriptor)
+
+    sink_dataquanta = \
+        plan.source("../test/lines.txt") \
+            .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_junction(descriptor):
+
+    plan = DataQuantaBuilder(descriptor)
+
+    dq_source_a = plan.source("../test/lines.txt")
+    dq_source_b = plan.source("../test/morelines.txt") \
+        .filter(lambda elem: str(elem).startswith("I"))
+    dq_source_c = plan.source("../test/lastlines.txt") \
+        .filter(lambda elem: str(elem).startswith("W"))
+
+    sink_dataquanta = dq_source_a.union(dq_source_b) \
+        .union(dq_source_c) \
+        .sort(lambda elem: elem.lower()) \
+        .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+def plan_java_junction(descriptor):
+
+    plan = DataQuantaBuilder(descriptor)
+
+    dq_source_a = plan.source("../test/lines.txt")
+    dq_source_b = plan.source("../test/morelines.txt")
+    sink_dataquanta = dq_source_a.union(dq_source_b) \
+        .filter(lambda elem: str(elem).startswith("I")) \
+        .sort(lambda elem: elem.lower()) \
+        .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+def plan_tpch_q1(descriptor):
+
+    #TODO create reduce by
+    plan = DataQuantaBuilder(descriptor)
+
+    def reducer(obj1, obj2):
+        return obj1[0]
+
+    sink = plan.source("../test/lineitem.txt") \
+        .map(lambda elem: elem.split("|")) \
+        .filter(lambda elem: datetime.datetime.strptime(elem[10], '%Y-%m-%d') <= datetime.datetime.strptime("1998-09-02", '%Y-%m-%d')) \
+        .map(lambda elem:
+           [elem[8], elem[9], elem[4], elem[5],
+            float(elem[5]) * (1 - float(elem[6])),
+            float(elem[5]) * (1 - float(elem[6])) * (1 + float(elem[7])),
+            elem[4], elem[5],
+            elem[6], 1]) \
+        .sink("../test/output.txt", end="")
+        # .group_by(lambda elem: elem) \
+        # .reduce_by(reducer) \
+        # .flatmap(lambda elem: elem.split("|"))
+        # .map(lambda elem: (elem, elem.split("|"))) \
+        # L_RETURNFLAG 8
+        # L_LINESTATUS 9
+        # L_QUANTITY 4
+        # L_EXTENDEDPRICE 5
+        # discount 6
+        # tax 7
+
+    return dq_source_b
+
+
+def plan_full_java(descriptor):
+
+    plan = DataQuantaBuilder(descriptor)
+
+    dq_source_a = plan.source("../test/lines.txt")
+    dq_source_b = plan.source("../test/morelines.txt")
+    sink_dataquanta = dq_source_a.union(dq_source_b) \
+        .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+if __name__ == '__main__':
+
+    # Plan will contain general info about the Wayang Plan created here
+    descriptor = Descriptor()
+
+    plan_dataquanta_sink = plan_tpch_q1(descriptor)
+    plan_dataquanta_sink.execute()
diff --git a/pywayang/orchestrator/main.py b/pywayang/orchestrator/main.py
new file mode 100644
index 0000000..b634eeb
--- /dev/null
+++ b/pywayang/orchestrator/main.py
@@ -0,0 +1,173 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from orchestrator.plan import Descriptor
+from orchestrator.dataquanta import DataQuantaBuilder
+import datetime
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_sort(descriptor):
+    plan = DataQuantaBuilder(descriptor)
+    sink_dataquanta = \
+        plan.source("../test/words.txt") \
+            .sort(lambda elem: elem.lower()) \
+            .sink("../test/output.txt", end="")
+    return sink_dataquanta
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_sort_filter(descriptor):
+    plan = DataQuantaBuilder(descriptor)
+    sink_dataquanta = \
+        plan.source("../test/words.txt") \
+            .sort(lambda elem: elem.lower()) \
+            .filter(lambda elem: str(elem).startswith("f")) \
+            .sink("../test/output.txt", end="")
+    return sink_dataquanta
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_filter_text(descriptor):
+    plan = DataQuantaBuilder(descriptor)
+
+    sink_dataquanta = \
+        plan.source("../test/words.txt") \
+            .filter(lambda elem: str(elem).startswith("f")) \
+            .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_filter(descriptor):
+    plan = DataQuantaBuilder(descriptor)
+
+    sink_dataquanta = \
+        plan.source("../test/numbers.txt") \
+            .filter(lambda elem: int(elem) % 2 != 0) \
+            .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_basic(descriptor):
+    plan = DataQuantaBuilder(descriptor)
+
+    sink_dataquanta = \
+        plan.source("../test/lines.txt") \
+            .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+# Returns the Sink Executable Dataquanta of a DEMO plan
+def plan_junction(descriptor):
+
+    plan = DataQuantaBuilder(descriptor)
+
+    dq_source_a = plan.source("../test/lines.txt")
+    dq_source_b = plan.source("../test/morelines.txt") \
+        .filter(lambda elem: str(elem).startswith("I"))
+    dq_source_c = plan.source("../test/lastlines.txt") \
+        .filter(lambda elem: str(elem).startswith("W"))
+
+    sink_dataquanta = dq_source_a.union(dq_source_b) \
+        .union(dq_source_c) \
+        .sort(lambda elem: elem.lower()) \
+        .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+def plan_java_junction(descriptor):
+
+    plan = DataQuantaBuilder(descriptor)
+
+    dq_source_a = plan.source("../test/lines.txt")
+    dq_source_b = plan.source("../test/morelines.txt")
+    sink_dataquanta = dq_source_a.union(dq_source_b) \
+        .filter(lambda elem: str(elem).startswith("I")) \
+        .sort(lambda elem: elem.lower()) \
+        .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+def plan_tpch_q1(descriptor):
+
+    # TODO create reduce by
+    plan = DataQuantaBuilder(descriptor)
+
+    def reducer(obj1, obj2):
+        return obj1[0], obj1[1], obj1[2] + obj2[2], obj1[3] + obj2[3], obj1[4] + obj2[4], obj1[5] + obj2[5], \
+               obj1[6] + obj2[6], obj1[7] + obj2[7], obj1[8] + obj2[8], obj1[9] + obj2[9]
+
+    sink = plan.source("../test/lineitem.txt") \
+        .map(lambda elem: elem.split("|")) \
+        .sink("../test/output.txt", end="")
+    """
+        .filter(lambda elem: datetime.datetime.strptime(elem[10], '%Y-%m-%d') <= datetime.datetime.strptime('1998-09-02', '%Y-%m-%d')) \
+        .map(lambda elem:
+             [elem[8], elem[9], elem[4], elem[5],
+              float(elem[5]) * (1 - float(elem[6])),
+              float(elem[5]) * (1 - float(elem[6])) * (1 + float(elem[7])),
+              elem[4], elem[5],
+              elem[6], 1]) \
+        .sink("../test/output.txt", end="")"""
+        # .reduce_by_key([0, 1], reducer) \
+
+
+    return sink
+
+
+def plan_full_java(descriptor):
+
+    plan = DataQuantaBuilder(descriptor)
+
+    dq_source_a = plan.source("../test/lines.txt")
+    dq_source_b = plan.source("../test/morelines.txt")
+    sink_dataquanta = dq_source_a.union(dq_source_b) \
+        .sink("../test/output.txt", end="")
+
+    return sink_dataquanta
+
+
+def plan_wordcount(descriptor):
+
+    plan = DataQuantaBuilder(descriptor)
+    sink_wordcount = plan.source("../test/lineitem.txt") \
+        .filter(lambda elem: len(str(elem).split("|")[0]) < 4) \
+        .flatmap(lambda elem: str(elem).split("|")) \
+        .sink("../test/output.txt", end="")
+
+    return sink_wordcount
+
+
+if __name__ == '__main__':
+
+    # Plan will contain general info about the Wayang Plan created here
+    descriptor = Descriptor()
+    descriptor.add_plugin(Descriptor.Plugin.spark)
+    descriptor.add_plugin(Descriptor.Plugin.java)
+
+    plan_dataquanta_sink = plan_wordcount(descriptor)
+    # plan_dataquanta_sink.execute()
+    # plan_dataquanta_sink.console()
+
+    plan_dataquanta_sink.to_wayang_plan()
diff --git a/pywayang/orchestrator/operator.py b/pywayang/orchestrator/operator.py
new file mode 100644
index 0000000..ecaa6bd
--- /dev/null
+++ b/pywayang/orchestrator/operator.py
@@ -0,0 +1,121 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import pickle
+import cloudpickle
+from config.config_reader import get_source_types
+from config.config_reader import get_sink_types
+from config.config_reader import get_boundary_types
+import logging
+
+pickle_protocol = pickle.HIGHEST_PROTOCOL
+
+
+# Describes an Operation over an intermediate result
+# Each operation could be processed by Python or Java platforms
+class Operator:
+
+    def __init__(
+            self, operator_type=None, udf=None, previous=None,
+            iterator=None, python_exec=False
+    ):
+
+        # Operator ID
+        self.id = id(self)
+
+        # Operator Type
+        self.operator_type = operator_type
+
+        # Set Boundaries
+        if self.operator_type in get_boundary_types():
+            self.boundary = True
+        else:
+            self.boundary = False
+
+        # UDF Function
+        self.udf = udf
+
+        # Source types must come with an Iterator
+        self.iterator = iterator
+        if operator_type in get_source_types():
+            if iterator is None:
+                print("Source Operator Type without an Iterator")
+                raise
+            else:
+                self.source = True
+        else:
+            self.source = False
+
+        # Sink Operators
+        if operator_type in get_sink_types():
+            self.sink = True
+        else:
+            self.sink = False
+
+        # TODO Why managing previous and predecessors per separate?
+        self.previous = previous
+
+        self.successor = []
+        self.predecessor = []
+
+        self.parameters = {}
+
+        # Set predecessors and successors from previous
+        if self.previous:
+            for prev in self.previous:
+                if prev is not None:
+                    prev.set_successor(self)
+                    self.set_predecessor(prev)
+
+        self.python_exec = python_exec
+
+        logging.info("Operator:" + str(self.getID()) + ", type:" + self.operator_type + ", PythonExecutable: " +
+                     str(self.python_exec) +
+                     ", is boundary: " + str(self.is_boundary()) + ", is source: " +
+                     str(self.source) + ", is sink: " + str(self.sink))
+
+    def getID(self):
+        return self.id
+
+    def is_source(self):
+        return self.source
+
+    def is_sink(self):
+        return self.sink
+
+    def is_boundary(self):
+        return self.boundary
+
+    def serialize_udf(self):
+        self.udf = cloudpickle.dumps(self.udf)
+
+    def getIterator(self):
+        if self.is_source():
+            return self.iterator
+        # TODO this should iterate through previous REDESIGN
+        return self.udf(self.previous[0].getIterator())
+
+    def set_parameter(self, key, value):
+        self.parameters[key] = value
+
+    def set_successor(self, suc):
+        if (not self.is_sink()) and self.successor.count(suc) == 0:
+            self.successor.append(suc)
+
+    def set_predecessor(self, suc):
+        if self.predecessor.count(suc) == 0:
+            self.predecessor.append(suc)
diff --git a/pywayang/orchestrator/plan.py b/pywayang/orchestrator/plan.py
new file mode 100644
index 0000000..25610cc
--- /dev/null
+++ b/pywayang/orchestrator/plan.py
@@ -0,0 +1,52 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import logging
+from enum import Enum
+
+class Descriptor:
+
+    def __init__(self):
+        self.sinks = []
+        self.sources = []
+        self.boundary_operators = None
+        logging.basicConfig(filename='../config/execution.log', level=logging.DEBUG)
+        self.plugins = []
+
+    class Plugin(Enum):
+        java = 0
+        spark = 1
+
+    def get_boundary_operators(self):
+        return self.boundary_operators
+
+    def add_source(self, operator):
+        self.sources.append(operator)
+
+    def get_sources(self):
+        return self.sources
+
+    def add_sink(self, operator):
+        self.sinks.append(operator)
+
+    def get_sinks(self):
+        return self.sinks
+
+    def add_plugin(self, plugin):
+        self.plugins.append(plugin)
+
+    def get_plugins(self):
+        return self.plugins
diff --git a/pywayang/protobuf/__init__.py b/pywayang/protobuf/__init__.py
new file mode 100644
index 0000000..15a80ad
--- /dev/null
+++ b/pywayang/protobuf/__init__.py
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import protobuf.pywayangplan_pb2
diff --git a/pywayang/protobuf/old_planwriter.py b/pywayang/protobuf/old_planwriter.py
new file mode 100644
index 0000000..e8700f0
--- /dev/null
+++ b/pywayang/protobuf/old_planwriter.py
@@ -0,0 +1,308 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import protobuf.pywayangplan_pb2 as pwb
+import os
+import pickle
+import struct
+import base64
+
+
+class OldMessageWriter:
+
+    def __init__(self, descriptor):
+
+        sink = descriptor.get_sinks()[0]
+        source = descriptor.get_sources()[0]
+
+        op = source
+        visited = []
+        middle_operators = []
+        while op.sink is not True and len(op.successor) > 0:
+            pre = op.successor[0]
+            if pre not in visited and pre.sink is not True:
+                pre.serialize_udf()
+                middle_operators.append(pre)
+                """base64_bytes = base64.b64encode(pre.udf)
+                pre.udf = base64_bytes"""
+
+                """pre.serialize_udf()
+                print("pre.udf")
+                print(pre.udf)
+                func = pickle.loads(pre.udf)
+                print("func")
+                print(func)
+                middle_operators.append(pre)
+
+                # Testing
+                msg = pre.udf
+                base64_bytes = base64.b64encode(msg)
+                base64_message = base64.b64decode(base64_bytes)
+                func2 = pickle.loads(base64_message)
+                print(base64_message)
+                func3 = pickle.loads(b'\x80\x04\x955\x04\x00\x00\x00\x00\x00\x00\x8c\x17cloudpickle.cloudpickle\x94\x8c\r_builtin_type\x94\x93\x94\x8c\nLambdaType\x94\x85\x94R\x94(h\x02\x8c\x08CodeType\x94\x85\x94R\x94(K\x01K\x00K\x01K\x03K\x13C\nt\x00\x88\x00|\x00\x83\x02S\x00\x94N\x85\x94\x8c\x06filter\x94\x85\x94\x8c\x08iterator\x94\x85\x94\x8cS/Users/rodrigopardomeza/wayang/incubator-wayang/pywayang/orchestrator/dataquanta.py\x94\x8c\x04func\x94K%C\x02\x00\x01\x94\x8c\x03udf\x94\x85\x94)t\x94R\x94}\x94(\x8c\x0b__package__\x94\x8c\x0corchestrator\x94\x8c\x08__name__\x94\x8c\x17orchestrator.dataquanta\x94\x8c\x08__file__\x94\x8cS/Users/rodrigopardomeza/wayang/incubator-wayang/pywayang/orchestrator/dataquanta.py\x94uNNh\x00\x8c\x10_make_empty_cell\x94\x93\x94)R\x94\x85\x94t\x94R\x94\x8c\x1ccloudpickle.cloudpickle_fast\x94\x8c\x12_function_setstate\x94\x93\x94h"}\x94}\x94(h\x19h\x10\x8c\x0c__qualname__\x94\x8c\x1fDataQuanta.filter.<locals>.func\x94\x8c\x0f__annotations__\x94}\x94\x8c\x0e__kwdefaults__\x94N\x8c\x0c__defaults__\x94N\x8c\n__module__\x94h\x1a\x8c\x07__doc__\x94N\x8c\x0b__closure__\x94h\x00\x8c\n_make_cell\x94\x93\x94h\x05(h\x08(K\x01K\x00K\x01K\x02KSC\x10t\x00|\x00\x83\x01d\x01\x16\x00d\x02k\x03S\x00\x94NK\x02K\x00\x87\x94\x8c\x03int\x94\x85\x94\x8c\x04elem\x94\x85\x94\x8cM/Users/rodrigopardomeza/wayang/incubator-wayang/pywayang/orchestrator/main.py\x94\x8c\x08<lambda>\x94K\x18C\x00\x94))t\x94R\x94}\x94(h\x17Nh\x19\x8c\x08__main__\x94h\x1b\x8cM/Users/rodrigopardomeza/wayang/incubator-wayang/pywayang/orchestrator/main.py\x94uNNNt\x94R\x94h%hB}\x94}\x94(h\x19h:h(\x8c\x1dplan_filter.<locals>.<lambda>\x94h*}\x94h,Nh-Nh.h?h/Nh0N\x8c\x17_cloudpickle_submodules\x94]\x94\x8c\x0b__globals__\x94}\x94u\x86\x94\x86R0\x85\x94R\x94\x85\x94hG]\x94hI}\x94u\x86\x94\x86R0.')
+                for i in func3([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]):
+                    print(i)"""
+            op = pre
+
+        """for mid in middle_operators:
+            print(mid.operator_type)
+            print(pickle.loads(mid.udf))
+            func = pickle.loads(mid.udf)
+            for i in func([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]):
+                print(i)"""
+
+        finalpath = "/Users/rodrigopardomeza/wayang/incubator-wayang/protobuf/filter_message"
+        planconf = pwb.WayangPlan()
+        try:
+            f = open(finalpath, "rb")
+            planconf.ParseFromString(f.read())
+            f.close()
+        except IOError:
+            print(finalpath + ": Could not open file.  Creating a new one.")
+
+        so = pwb.Source()
+        so.id = source.id
+        so.type = source.operator_type
+        so.path = os.path.abspath(source.udf)
+
+        operators = []
+        for mid in middle_operators:
+            op = pwb.Operator()
+            op.id = mid.id
+            op.type = mid.operator_type
+            op.udf = mid.udf
+            operators.append(op)
+
+        si = pwb.Sink()
+        si.id = sink.id
+        si.type = sink.operator_type
+        si.path = os.path.abspath(sink.udf)
+
+        plan = pwb.Plan()
+        plan.source.CopyFrom(so)
+        plan.sink.CopyFrom(si)
+        plan.operators.extend(operators)
+        plan.input = pwb.Plan.string
+        plan.output = pwb.Plan.string
+
+        ctx = pwb.Context()
+        ctx.platforms.extend([pwb.Context.Platform.java])
+
+        planconf.plan.CopyFrom(plan)
+        planconf.context.CopyFrom(ctx)
+
+        f = open(finalpath, "wb")
+        f.write(planconf.SerializeToString())
+        f.close()
+        pass
+
+class func_inteface:
+
+    def __init__(self, node, nested_udf):
+        self.node = node
+        self.nested_udf = nested_udf
+
+    def func(self, iterable):
+        return self.node.operator.udf(self.nested_udf(iterable))
+
+
+class MessageWriter:
+    sources = []
+    operators = []
+    sinks = []
+
+    def add_source(self, operator_id, operator_type, path, predecessors, successors):
+        source = pwb.OperatorProto()
+        source.id = operator_id
+        source.type = operator_type
+        source.path = os.path.abspath(path)
+        source.udf = None
+        source.predecessors = predecessors
+        source.successors = successors
+        self.sources.append(source)
+
+    def add_sink(self, operator_id, operator_type, path, predecessors, successors):
+        sink = pwb.OperatorProto()
+        sink.id = operator_id
+        sink.type = operator_type
+        sink.path = os.path.abspath(path)
+        sink.udf = None
+        sink.predecessors = predecessors
+        sink.successors = successors
+        self.sinks.append(sink)
+
+    def add_operator(self, operator_id, operator_type, udf, path, predecessors, successors):
+        op = pwb.OperatorProto()
+        op.id = operator_id
+        op.type = operator_type
+        op.udf = udf
+        op.path = path
+        op.predecessors = predecessors
+        op.successors = successors
+        self.operators.append(op)
+
+    def process_pipeline(self, stage):
+
+        nested_udf = None
+        nested_id = ""
+        for node in reversed(stage):
+            print("########")
+            print(node.operator_type, "executable:", node.python_exec, "id:", node.id)
+
+            if nested_udf is not None:
+                print("review pre")
+                print( nested_udf)
+                print( nested_udf(["Wilo","lifo","Wifo"]))
+
+            if not node.python_exec:
+                if nested_udf is not None:
+                    """self.add_operator(nested_id, "map_partition", nested_udf, None
+                                      # obtain predecessors and successors
+                                      , successors=[node.id]
+                                      )"""
+                    print("node", nested_id)
+                    print(nested_udf)
+                    print("he muerto")
+                    print( nested_udf(["Wilo","lifo","Wifo"]))
+
+                    t = nested_udf(["Wilo","lifo","Wifo"])
+                    print("jajajarvard")
+                    print(t)
+                    for i in t:
+                        print(i)
+                    nested_udf = None
+                    nested_id = ""
+
+                """if node.operator.source:
+                    self.add_source(
+                        node.id, node.operator_type, node.operator.udf,
+                        node.predecessors, node.operator.successor)
+                else:
+                    self.add_operator(
+                        node.id, node.operator_type, None, node.operator.udf,
+                        node.predecessors, node.operator.successor)"""
+            else:
+                print("adding", node.id)
+                if nested_udf is None:
+                    nested_udf = node.operator.udf
+                    nested_id = node.id
+                else:
+                    print("paseeeeeee viste")
+                    tmp = nested_udf
+
+                    print( tmp(["Wilo","lifo","Wifo"]))
+
+                    #def func(_, iterable):
+                    #    return nested_udf(node.operator.udf(iterable))
+                    nested_udf = self.concatenate(nested_udf, node.operator.udf)
+                    print( nested_udf(["Wilo","lifo","Wifo"]))
+                    print(nested_udf)
+
+                    # nested_udf = func_inteface(node, nested_udf)
+                    nested_id = str(node.id) + "," + str(nested_id)
+
+            if nested_udf is not None:
+                print("review")
+                print( nested_udf)
+                print( nested_udf(["Wilo","lifo","Wifo"]))
+
+        if nested_udf is not None:
+            """self.add_operator(nested_id, "map_partition", nested_udf, None
+                              # obtain predecessors and successors
+                              , successors=[node.id]
+                              )"""
+            print("node", nested_id)
+            print(nested_udf)
+            t = nested_udf(["Wilo","lifo","Wifo"])
+            print("jajajarvard2")
+            print(t)
+            for i in t:
+                print(i)
+            nested_udf = None
+            nested_id = ""
+
+    def __init__(self):
+        print("lala")
+
+    def concatenate(self, function_a, function_b):
+        def executable(iterable):
+            return function_a(function_b(iterable))
+        return executable
+
+    def old(self, descriptor):
+
+        sink = descriptor.get_sinks()[0]
+        source = descriptor.get_sources()[0]
+
+        op = source
+        visited = []
+        middle_operators = []
+        while op.sink is not True and len(op.successor) > 0:
+            pre = op.successor[0]
+            if pre not in visited and pre.sink is not True:
+                pre.serialize_udf()
+                middle_operators.append(pre)
+            op = pre
+
+        finalpath = "/Users/rodrigopardomeza/wayang/incubator-wayang/protobuf/filter_message"
+        planconf = pwb.WayangPlan()
+        try:
+            f = open(finalpath, "rb")
+            planconf.ParseFromString(f.read())
+            f.close()
+        except IOError:
+            print(finalpath + ": Could not open file.  Creating a new one.")
+
+        so = pwb.Source()
+        so.id = source.id
+        so.type = source.operator_type
+        so.path = os.path.abspath(source.udf)
+
+        operators = []
+        for mid in middle_operators:
+            op = pwb.Operator()
+            op.id = mid.id
+            op.type = mid.operator_type
+            op.udf = mid.udf
+            operators.append(op)
+
+        si = pwb.Sink()
+        si.id = sink.id
+        si.type = sink.operator_type
+        si.path = os.path.abspath(sink.udf)
+
+        plan = pwb.Plan()
+        plan.source.CopyFrom(so)
+        plan.sink.CopyFrom(si)
+        plan.operators.extend(operators)
+        plan.input = pwb.Plan.string
+        plan.output = pwb.Plan.string
+
+        ctx = pwb.Context()
+        ctx.platforms.extend([pwb.Context.Platform.java])
+
+        planconf.plan.CopyFrom(plan)
+        planconf.context.CopyFrom(ctx)
+
+        f = open(finalpath, "wb")
+        f.write(planconf.SerializeToString())
+        f.close()
+        pass
+
+    def pipeline_singleton(self):
+        print("lala")
diff --git a/pywayang/protobuf/planwriter.py b/pywayang/protobuf/planwriter.py
new file mode 100644
index 0000000..b63dcbb
--- /dev/null
+++ b/pywayang/protobuf/planwriter.py
@@ -0,0 +1,277 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import protobuf.pywayangplan_pb2 as pwb
+import os
+import cloudpickle
+import logging
+import pathlib
+import requests
+import base64
+
+
+# Writes Wayang Plan from several stages
+class MessageWriter:
+    sources = []
+    operators = []
+    sinks = []
+    operator_references = {}
+    boundaries = {}
+
+    # Creates and appends Source type of operator
+    def add_source(self, operator_id, operator_type, path):
+        source = pwb.OperatorProto()
+        source.id = str(operator_id)
+        source.type = operator_type
+        source.path = os.path.abspath(path)
+        source.udf = chr(0).encode('utf-8')
+        # source.parameters = {}
+        self.sources.append(source)
+        return source
+
+    # Creates and appends Sink type of operator
+    def add_sink(self, operator_id, operator_type, path):
+        sink = pwb.OperatorProto()
+        sink.id = str(operator_id)
+        sink.type = operator_type
+        sink.path = os.path.abspath(path)
+        sink.udf = chr(0).encode('utf-8')
+        # sink.parameters = {}
+        self.sinks.append(sink)
+        return sink
+
+    # Creates and appends a Python operator
+    # Python OP don't require parameters, UDF has the function ready to be executed directly
+    def add_operator(self, operator_id, operator_type, udf):
+        op = pwb.OperatorProto()
+        op.id = str(operator_id)
+        op.type = operator_type
+        op.udf = cloudpickle.dumps(udf)
+        op.path = str(None)
+        # op.parameters = {}
+        self.operators.append(op)
+        return op
+
+    # Creates and appends a Java operator
+    def add_java_operator(self, operator_id, operator_type, udf, parameters):
+        op = pwb.OperatorProto()
+        op.id = str(operator_id)
+        op.type = operator_type
+        op.udf = cloudpickle.dumps(udf)
+        op.path = str(None)
+        #op.parameters = parameters
+        for param in parameters:
+            print(param, parameters[param])
+            op.parameters[param] = str(parameters[param])
+            # op.parameters[]
+        #m.mapfield[5] = 10
+        self.operators.append(op)
+        return op
+
+    # Receive a chain of operators, separate them in Wayang Operators
+    # Compacts several Python executable operators in one Map Partition Wayang Operator
+    def process_pipeline(self, stage):
+
+        nested_udf = None
+        nested_id = ""
+        nested_predecessors = None
+        nested_successors = None
+        for node in reversed(stage):
+            logging.debug(node.operator_type + " executable: " + str(node.python_exec) + " id: " + str(node.id))
+
+            if not node.python_exec:
+                if nested_udf is not None:
+
+                    # Predecessors depends on last operator
+                    # Successors depends on first operator
+                    op = self.add_operator(nested_id, "map_partition", nested_udf)
+
+                    ids = str(nested_id).split(",")
+                    for id in ids:
+                        self.operator_references[str(id)] = op
+
+                    self.boundaries[str(nested_id)] = {}
+                    self.boundaries[str(nested_id)]["end"] = nested_successors
+                    self.boundaries[str(nested_id)]["start"] = nested_predecessors
+
+                    nested_udf = None
+                    nested_id = ""
+                    nested_predecessors = None
+                    nested_successors = None
+
+                if node.operator.source:
+                    op = self.add_source(node.id, node.operator_type, node.operator.udf)
+                    self.operator_references[str(node.id)] = op
+                    self.boundaries[str(node.id)] = {}
+                    self.boundaries[str(node.id)]["end"] = node.successors.keys()
+
+                elif node.operator.sink:
+                    op = self.add_sink(node.id, node.operator_type, node.operator.udf)
+                    self.operator_references[str(node.id)] = op
+                    self.boundaries[str(node.id)] = {}
+                    self.boundaries[str(node.id)]["start"] = node.predecessors.keys()
+
+                # Regular operator to be processed in Java
+                # Notice that those could include more parameters for Java
+                else:
+                    op = self.add_java_operator(node.id, node.operator_type, node.operator.udf, node.operator.parameters)
+                    self.operator_references[str(node.id)] = op
+                    self.boundaries[str(node.id)] = {}
+                    self.boundaries[str(node.id)]["start"] = node.predecessors.keys()
+                    self.boundaries[str(node.id)]["end"] = node.successors.keys()
+
+            else:
+
+                if nested_udf is None:
+                    nested_udf = node.operator.udf
+                    nested_id = node.id
+                    # It is the last operator to execute in the map partition
+                    nested_successors = node.successors.keys()
+
+                else:
+                    nested_udf = self.concatenate(nested_udf, node.operator.udf)
+                    nested_id = str(node.id) + "," + str(nested_id)
+
+                # Every iteration assign the first known predecessors
+                nested_predecessors = node.predecessors.keys()
+
+        # Just in case in the future some pipelines start with Python operators
+        if nested_udf is not None:
+            self.add_operator(nested_id, "map_partition", nested_udf)
+
+            ids = nested_id.split(",")
+            for id in ids:
+                self.operator_references[id] = op
+
+            self.boundaries[nested_id] = {}
+            self.boundaries[nested_id]["end"] = nested_successors
+            self.boundaries[nested_id]["start"] = nested_predecessors
+
+    def __init__(self):
+        pass
+
+    # Takes 2 Functions and compact them in only one function
+    @staticmethod
+    def concatenate(function_a, function_b):
+        def executable(iterable):
+            return function_a(function_b(iterable))
+
+        return executable
+
+    # Set dependencies over final Wayang Operators
+    def set_dependencies(self):
+
+        for source in self.sources:
+
+            if 'end' in self.boundaries[source.id]:
+                op_successors = []
+                for op_id in self.boundaries[source.id]['end']:
+                    op_successors.append(str(self.operator_references[str(op_id)].id))
+                source.successors.extend(op_successors)
+
+        for sink in self.sinks:
+            if 'start' in self.boundaries[sink.id]:
+                op_predecessors = []
+                for op_id in self.boundaries[sink.id]['start']:
+                    op_predecessors.append(str(self.operator_references[str(op_id)].id))
+                sink.predecessors.extend(op_predecessors)
+
+        for op in self.operators:
+            if 'start' in self.boundaries[op.id]:
+                op_predecessors = []
+                for op_id in self.boundaries[op.id]['start']:
+                    op_predecessors.append(str(self.operator_references[str(op_id)].id))
+                op.predecessors.extend(op_predecessors)
+
+            if 'end' in self.boundaries[op.id]:
+                op_successors = []
+                for op_id in self.boundaries[op.id]['end']:
+                    op_successors.append(str(self.operator_references[str(op_id)].id))
+                op.successors.extend(op_successors)
+
+    # Writes the message to a local directory
+    def write_message(self, descriptor):
+
+        finalpath = "../../protobuf/wayang_message"
+        plan_configuration = pwb.WayangPlanProto()
+
+        try:
+            f = open(finalpath, "rb")
+            plan_configuration.ParseFromString(f.read())
+            f.close()
+        except IOError:
+            logging.warn("File " + finalpath + " did not exist. System generated a new file")
+
+        plan = pwb.PlanProto()
+        plan.sources.extend(self.sources)
+        plan.operators.extend(self.operators)
+        plan.sinks.extend(self.sinks)
+        plan.input = pwb.PlanProto.string
+        plan.output = pwb.PlanProto.string
+
+        ctx = pwb.ContextProto()
+        # ctx.platforms.extend([pwb.ContextProto.PlatformProto.java])
+        for plug in descriptor.plugins:
+            ctx.platforms.append(plug.value)
+        # ctx.platforms.extend(descriptor.get_plugins())
+
+        plan_configuration.plan.CopyFrom(plan)
+        plan_configuration.context.CopyFrom(ctx)
+
+        f = open(finalpath, "wb")
+        f.write(plan_configuration.SerializeToString())
+        f.close()
+        pass
+
+    # Send message as bytes to the Wayang Rest API
+    def send_message(self, descriptor):
+
+        plan_configuration = pwb.WayangPlanProto()
+
+        plan = pwb.PlanProto()
+        plan.sources.extend(self.sources)
+        plan.operators.extend(self.operators)
+        plan.sinks.extend(self.sinks)
+        plan.input = pwb.PlanProto.string
+        plan.output = pwb.PlanProto.string
+
+        ctx = pwb.ContextProto()
+        # ctx.platforms.extend([pwb.ContextProto.PlatformProto.java])
+        for plug in descriptor.plugins:
+            ctx.platforms.append(plug.value)
+        # ctx.platforms.extend(descriptor.get_plugins())
+
+        plan_configuration.plan.CopyFrom(plan)
+        plan_configuration.context.CopyFrom(ctx)
+
+        print("plan!")
+        print(plan_configuration)
+
+        msg_bytes = plan_configuration.SerializeToString()
+        msg_64 = base64.b64encode(msg_bytes)
+
+        logging.debug(msg_bytes)
+        # response = requests.get("http://localhost:8080/plan/create/fromfile")
+        data = {
+            'message': msg_64
+        }
+        response = requests.post("http://localhost:8080/plan/create", data)
+        logging.debug(response)
+        # f = open(finalpath, "wb")
+        # f.write(plan_configuration.SerializeToString())
+        # f.close()
+        pass
diff --git a/pywayang/test/demo_testing.py b/pywayang/test/demo_testing.py
new file mode 100644
index 0000000..c096a89
--- /dev/null
+++ b/pywayang/test/demo_testing.py
@@ -0,0 +1,30 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import unittest
+
+
+class MyTestCase(unittest.TestCase):
+
+    def test_something(self):
+        self.assertEqual(True, False)
+
+    def test_upper(self):
+        self.assertEqual('foo'.upper(), 'FOO')
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/pywayang/test/full_java_test.py b/pywayang/test/full_java_test.py
new file mode 100644
index 0000000..d17aedd
--- /dev/null
+++ b/pywayang/test/full_java_test.py
@@ -0,0 +1,69 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import unittest
+from orchestrator.plan import Descriptor
+from orchestrator.dataquanta import DataQuantaBuilder
+
+
+class MyTestCase(unittest.TestCase):
+
+    def test_most_basic(self):
+        descriptor = Descriptor()
+        descriptor.add_plugin(Descriptor.Plugin.java)
+
+        plan = DataQuantaBuilder(descriptor)
+        sink_dataquanta = \
+            plan.source("../test/lines.txt") \
+                .sink("../test/output.txt", end="")
+
+        sink_dataquanta.to_wayang_plan()
+
+
+    def test_single_juncture(self):
+        descriptor = Descriptor()
+        descriptor.add_plugin(Descriptor.Plugin.java)
+
+        plan = DataQuantaBuilder(descriptor)
+        dq_source_a = plan.source("../test/lines.txt")
+        dq_source_b = plan.source("../test/morelines.txt")
+        sink_dataquanta = dq_source_a.union(dq_source_b) \
+            .sink("../test/output.txt", end="")
+
+        sink_dataquanta.to_wayang_plan()
+
+
+    def test_multiple_juncture(self):
+        descriptor = Descriptor()
+        descriptor.add_plugin(Descriptor.Plugin.java)
+
+        plan = DataQuantaBuilder(descriptor)
+        dq_source_a = plan.source("../test/lines.txt")
+        dq_source_b = plan.source("../test/morelines.txt") \
+            .filter(lambda elem: str(elem).startswith("I"))
+        dq_source_c = plan.source("../test/lastlines.txt") \
+            .filter(lambda elem: str(elem).startswith("W"))
+
+        sink_dataquanta = dq_source_a.union(dq_source_b) \
+            .union(dq_source_c) \
+            .sort(lambda elem: elem.lower()) \
+            .sink("../test/output.txt", end="")
+
+        sink_dataquanta.to_wayang_plan()
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/pywayang/test/full_spark_test.py b/pywayang/test/full_spark_test.py
new file mode 100644
index 0000000..9276ccc
--- /dev/null
+++ b/pywayang/test/full_spark_test.py
@@ -0,0 +1,67 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import unittest
+from orchestrator.plan import Descriptor
+from orchestrator.dataquanta import DataQuantaBuilder
+
+
+def test_most_basic(self):
+    descriptor = Descriptor()
+    descriptor.add_plugin(Descriptor.Plugin.spark)
+
+    plan = DataQuantaBuilder(descriptor)
+    sink_dataquanta = \
+        plan.source("../test/lines.txt") \
+            .sink("../test/output.txt", end="")
+
+    sink_dataquanta.to_wayang_plan()
+
+
+def test_single_juncture(self):
+    descriptor = Descriptor()
+    descriptor.add_plugin(Descriptor.Plugin.spark)
+
+    plan = DataQuantaBuilder(descriptor)
+    dq_source_a = plan.source("../test/lines.txt")
+    dq_source_b = plan.source("../test/morelines.txt")
+    sink_dataquanta = dq_source_a.union(dq_source_b) \
+        .sink("../test/output.txt", end="")
+
+    sink_dataquanta.to_wayang_plan()
+
+
+def test_multiple_juncture(self):
+    descriptor = Descriptor()
+    descriptor.add_plugin(Descriptor.Plugin.spark)
+
+    plan = DataQuantaBuilder(descriptor)
+    dq_source_a = plan.source("../test/lines.txt")
+    dq_source_b = plan.source("../test/morelines.txt") \
+        .filter(lambda elem: str(elem).startswith("I"))
+    dq_source_c = plan.source("../test/lastlines.txt") \
+        .filter(lambda elem: str(elem).startswith("W"))
+
+    sink_dataquanta = dq_source_a.union(dq_source_b) \
+        .union(dq_source_c) \
+        .sort(lambda elem: elem.lower()) \
+        .sink("../test/output.txt", end="")
+
+    sink_dataquanta.to_wayang_plan()
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/wayang-api/pom.xml b/wayang-api/pom.xml
index e31751b..c868801 100644
--- a/wayang-api/pom.xml
+++ b/wayang-api/pom.xml
@@ -38,6 +38,7 @@
     <modules>
         <module>wayang-api-scala-java</module>
         <module>wayang-api-python</module>
+        <module>wayang-api-rest</module>
     </modules>
 
 </project>
diff --git a/wayang-api/wayang-api-python/pom.xml b/wayang-api/wayang-api-python/pom.xml
index 5a3ad44..647f07e 100644
--- a/wayang-api/wayang-api-python/pom.xml
+++ b/wayang-api/wayang-api-python/pom.xml
@@ -30,9 +30,54 @@
     <version>0.6.1-SNAPSHOT</version>
 
     <name>Wayang API Python</name>
-    <description>Wayang implementation of an API of Python to be enable to work with code writed in python</description>
+    <description>Wayang implementation of an API of Python to be enable to work with code wrote in python</description>
 
     <properties>
         <java-module-name>org.apache.wayang.api</java-module-name>
+        <graphchi.version>0.2.2</graphchi.version>
+        <spark.version>2.4.0</spark.version>
+        <flink.version>1.7.1</flink.version>
+        <scala.mayor.version>2.11</scala.mayor.version>
+        <giraph.version>1.2.0-hadoop2</giraph.version>
     </properties>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.apache.wayang</groupId>
+                <artifactId>wayang-commons</artifactId>
+                <version>0.6.1-SNAPSHOT</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-core</artifactId>
+            <version>0.6.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-java</artifactId>
+            <version>0.6.1-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+            <version>3.7.0</version>
+            <scope>compile</scope>
+        </dependency>
+    </dependencies>
+
 </project>
diff --git a/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/ProcessFeeder.java b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/ProcessFeeder.java
new file mode 100644
index 0000000..b463b36
--- /dev/null
+++ b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/ProcessFeeder.java
@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.python.executor;
+
+import com.google.protobuf.ByteString;
+import org.apache.wayang.api.python.function.PythonUdf;
+import org.apache.wayang.core.api.exception.WayangException;
+
+import java.io.BufferedOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.net.Socket;
+import java.net.SocketException;
+import java.nio.charset.StandardCharsets;
+import java.util.Iterator;
+import java.util.Map;
+
+public class ProcessFeeder<Input, Output> {
+
+    private Socket socket;
+    private PythonUdf<Input, Output> udf;
+    private ByteString serializedUDF;
+    private Iterable<Input> input;
+
+    //TODO add to a config file
+    int END_OF_DATA_SECTION = -1;
+    int NULL = -5;
+
+    public ProcessFeeder(
+            Socket socket,
+            PythonUdf<Input, Output> udf,
+            ByteString serializedUDF,
+            Iterable<Input> input){
+
+        if(input == null) throw new WayangException("Nothing to process with Python API");
+
+        this.socket = socket;
+        this.udf = udf;
+        this.serializedUDF = serializedUDF;
+        this.input = input;
+
+    }
+
+    public void send(){
+
+        try{
+            //TODO use config buffer size
+            int BUFFER_SIZE = 8192;
+
+            BufferedOutputStream stream = new BufferedOutputStream(socket.getOutputStream(), BUFFER_SIZE);
+            DataOutputStream dataOut = new DataOutputStream(stream);
+
+            writeUDF(serializedUDF, dataOut);
+            this.writeIteratorToStream(input.iterator(), dataOut);
+            dataOut.writeInt(END_OF_DATA_SECTION);
+            dataOut.flush();
+
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public void writeUDF(ByteString serializedUDF, DataOutputStream dataOut){
+
+        //write(serializedUDF.toByteArray(), dataOut);
+        writeBytes(serializedUDF.toByteArray(), dataOut);
+        System.out.println("UDF written");
+
+    }
+
+    public void writeIteratorToStream(Iterator<Input> iter, DataOutputStream dataOut){
+
+        System.out.println("iterator being send");
+        for (Iterator<Input> it = iter; it.hasNext(); ) {
+            Input elem = it.next();
+            //System.out.println(elem.toString());
+            write(elem, dataOut);
+        }
+    }
+
+    /*TODO Missing case PortableDataStream */
+    public void write(Object obj, DataOutputStream dataOut){
+        try {
+
+            if(obj == null)
+                dataOut.writeInt(this.NULL);
+
+            /**
+             * Byte Array cases
+             */
+            else if (obj instanceof Byte[] || obj instanceof byte[]) {
+                System.out.println("Writing Bytes");
+                writeBytes(obj, dataOut);
+            }
+            /**
+             * String case
+             * */
+            else if (obj instanceof String)
+                writeUTF((String) obj, dataOut);
+
+            /**
+             * Key, Value case
+             * */
+            else if (obj instanceof Map.Entry)
+                writeKeyValue((Map.Entry) obj, dataOut);
+
+            else{
+                throw new WayangException("Unexpected element type " + obj.getClass());
+            }
+
+
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public void writeBytes(Object obj, DataOutputStream dataOut){
+
+        try{
+
+            if (obj instanceof Byte[]) {
+
+                int length = ((Byte[]) obj).length;
+
+                byte[] bytes = new byte[length];
+                int j=0;
+
+                // Unboxing Byte values. (Byte[] to byte[])
+                for(Byte b: ((Byte[]) obj))
+                    bytes[j++] = b.byteValue();
+
+                dataOut.writeInt(length);
+                dataOut.write(bytes);
+
+            } else if (obj instanceof byte[]) {
+
+                dataOut.writeInt(((byte[]) obj).length);
+                dataOut.write(((byte[]) obj));
+            }
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public void writeUTF(String str, DataOutputStream dataOut){
+
+        byte[] bytes = str.getBytes(StandardCharsets.UTF_8);
+
+        try {
+
+            dataOut.writeInt(bytes.length);
+            dataOut.write(bytes);
+        } catch (SocketException e){
+
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public void writeKeyValue(Map.Entry obj, DataOutputStream dataOut){
+
+        write(obj.getKey(), dataOut);
+        write(obj.getValue(), dataOut);
+    }
+
+}
diff --git a/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/ProcessReceiver.java b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/ProcessReceiver.java
new file mode 100644
index 0000000..895245a
--- /dev/null
+++ b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/ProcessReceiver.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.python.executor;
+
+import java.io.BufferedInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.net.Socket;
+import java.util.Iterator;
+/*TODO cannot be always string, include definition for every operator
+*  like: map(udf, inputtype, outputtype)*/
+public class ProcessReceiver<Output> {
+
+    private ReaderIterator<Output> iterator;
+
+    public ProcessReceiver(Socket socket){
+        try{
+            //TODO use config buffer size
+            int BUFFER_SIZE = 8192;
+
+            DataInputStream stream = new DataInputStream(new BufferedInputStream(socket.getInputStream(), BUFFER_SIZE));
+            this.iterator = new ReaderIterator<>(stream);
+
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public Iterable<Output> getIterable(){
+        return () -> iterator;
+    }
+
+    public void print(){
+        iterator.forEachRemaining(x -> System.out.println(x.toString()));
+
+    }
+}
diff --git a/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/PythonProcessCaller.java b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/PythonProcessCaller.java
new file mode 100644
index 0000000..0d0e0b9
--- /dev/null
+++ b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/PythonProcessCaller.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.python.executor;
+
+import com.google.protobuf.ByteString;
+import org.apache.wayang.core.api.Configuration;
+import org.apache.wayang.core.api.exception.WayangException;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.Map;
+
+public class PythonProcessCaller {
+
+    private Process process;
+    private Socket socket;
+    private ServerSocket serverSocket;
+    private boolean ready;
+
+    //TODO How to get the config
+    private Configuration configuration;
+
+    public PythonProcessCaller(ByteString serializedUDF){
+
+        //TODO create documentation to how to the configuration in the code
+        this.configuration = new Configuration("file:///Users/rodrigopardomeza/wayang/incubator-wayang/wayang-api/wayang-api-python/src/main/resources/wayang-api-python-defaults.properties");
+        this.ready = false;
+        byte[] addr = new byte[4];
+        addr[0] = 127; addr[1] = 0; addr[2] = 0; addr[3] = 1;
+
+        try {
+            /*TODO should NOT be assigned an specific port, set port as 0 (zero)*/
+            this.serverSocket = new ServerSocket(0, 1, InetAddress.getByAddress(addr));
+            ProcessBuilder pb = new ProcessBuilder(
+                    Arrays.asList(
+                            "python3",
+                            this.configuration.getStringProperty("wayang.api.python.worker")
+                    )
+            );
+            Map<String, String> workerEnv = pb.environment();
+            workerEnv.put("PYTHON_WORKER_FACTORY_PORT", String.valueOf(this.serverSocket.getLocalPort()));
+
+            // TODO See what is happening with ENV Python version
+            workerEnv.put("PYTHONPATH", "/Users/rodrigopardomeza/wayang/incubator-wayang/pywayang/:/Users/rodrigopardomeza/opt/anaconda3/");
+
+            pb.redirectOutput(ProcessBuilder.Redirect.INHERIT);
+            pb.redirectError(ProcessBuilder.Redirect.INHERIT);
+            this.process = pb.start();
+
+
+            // Redirect worker stdout and stderr
+            //IDK redirectStreamsToStderr(worker.getInputStream, worker.getErrorStream)
+
+            // Wait for it to connect to our socket
+            this.serverSocket.setSoTimeout(10000);
+
+            try {
+                this.socket = this.serverSocket.accept();
+                this.serverSocket.setSoTimeout(0);
+
+                if(socket.isConnected())
+                    this.ready = true;
+
+            } catch (Exception e) {
+                System.out.println(e);
+                throw new WayangException("Python worker failed to connect back.", e);
+            }
+        } catch (Exception e){
+            System.out.println(e);
+            throw new WayangException("Python worker failed");
+        }
+    }
+
+    public Process getProcess() {
+        return process;
+    }
+
+    public Socket getSocket() {
+        return socket;
+    }
+
+    public boolean isReady(){
+        return ready;
+    }
+
+    public void close(){
+        try {
+            this.process.destroy();
+            this.socket.close();
+            this.serverSocket.close();
+            System.out.println("Everything closed");
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+}
diff --git a/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/PythonWorkerManager.java b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/PythonWorkerManager.java
new file mode 100644
index 0000000..09b394d
--- /dev/null
+++ b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/PythonWorkerManager.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.python.executor;
+
+import com.google.protobuf.ByteString;
+import org.apache.wayang.api.python.function.PythonUdf;
+import org.apache.wayang.core.api.exception.WayangException;
+
+public class PythonWorkerManager<Input, Output> {
+
+    private PythonUdf<Input, Output> udf;
+    private ByteString serializedUDF;
+    private Iterable<Input> inputIterator;
+
+    public PythonWorkerManager(
+            PythonUdf<Input, Output> udf,
+            ByteString serializedUDF,
+            Iterable<Input> input
+    ){
+        this.udf = udf;
+        this.serializedUDF = serializedUDF;
+        this.inputIterator = input;
+    }
+
+    public Iterable<Output> execute(){
+        PythonProcessCaller worker = new PythonProcessCaller(this.serializedUDF);
+
+        if(worker.isReady()){
+
+            ProcessFeeder<Input, Output> feed = new ProcessFeeder<>(
+                    worker.getSocket(),
+                    this.udf,
+                    this.serializedUDF,
+                    this.inputIterator
+            );
+            feed.send();
+            ProcessReceiver<Output> r = new ProcessReceiver<>(worker.getSocket());
+
+            //r.print();
+            return r.getIterable();
+            //return (Iterable<Output>) this.inputIterator;
+
+        } else{
+
+            int port = worker.getSocket().getLocalPort();
+            worker.close();
+            throw new WayangException("Not possible to work with the Socket provided on port: " + port);
+        }
+
+    }
+}
diff --git a/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/ReaderIterator.java b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/ReaderIterator.java
new file mode 100644
index 0000000..1f90358
--- /dev/null
+++ b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/executor/ReaderIterator.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.python.executor;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+public class ReaderIterator <Output> implements Iterator<Output> {
+
+    private Output nextObj = null;
+    private boolean eos = false;
+    private boolean fst = false;
+    private DataInputStream stream = null;
+
+    public ReaderIterator(DataInputStream stream) {
+
+        this.stream = stream;
+        this.eos = false;
+        this.nextObj = null;
+    }
+
+    private Output read() {
+
+        int END_OF_DATA_SECTION = -1;
+
+        try {
+            int length = this.stream.readInt();
+
+            if (length > 0) {
+                byte[] obj = new byte[length];
+                stream.readFully(obj);
+                String s = new String(obj, StandardCharsets.UTF_8);
+                Output it = (Output) s;
+                return it;
+            } else if (length == END_OF_DATA_SECTION) {
+                this.eos = true;
+                return null;
+            }
+        } catch (IOException e) {
+            //e.printStackTrace();
+            throw new RuntimeException(e);
+        }
+        return null;
+    }
+
+    @Override
+    public boolean hasNext() {
+
+        if(!this.eos){
+            nextObj = read();
+            System.out.println(nextObj + " " + !this.eos);
+
+            /*To work with null values it is suppose to use -5
+            if(this.nextObj == null){
+                return false;
+            }*/
+
+            return !this.eos;
+        }
+
+        return false;
+    }
+
+    @Override
+    public Output next() {
+
+        if(!this.eos){
+            Output obj = nextObj;
+            nextObj = null;
+            return obj;
+        }
+
+        throw new NoSuchElementException();
+    }
+
+}
diff --git a/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/function/PythonUdf.java b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/function/PythonUdf.java
new file mode 100644
index 0000000..88134b3
--- /dev/null
+++ b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/function/PythonUdf.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.python.function;
+
+import org.apache.wayang.core.function.FunctionDescriptor;
+
+public interface PythonUdf<Input, Output> extends FunctionDescriptor.SerializableFunction<Iterable<Input>, Iterable<Output>>{
+
+}
diff --git a/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/function/WrappedPythonFunction.java b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/function/WrappedPythonFunction.java
new file mode 100644
index 0000000..26382f4
--- /dev/null
+++ b/wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/function/WrappedPythonFunction.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.python.function;
+
+import com.google.protobuf.ByteString;
+import org.apache.wayang.api.python.executor.PythonWorkerManager;
+import org.apache.wayang.core.function.FunctionDescriptor;
+
+public class WrappedPythonFunction<Input, Output>  implements FunctionDescriptor.SerializableFunction<Iterable<Input>, Iterable<Output>> {
+
+    private PythonUdf<Input, Output> myUDF;
+    private ByteString serializedUDF;
+
+    public WrappedPythonFunction(PythonUdf<Input, Output> myUDF, ByteString serializedUDF){
+        this.myUDF = myUDF;
+        this.serializedUDF = serializedUDF;
+    }
+
+    @Override
+    public Iterable<Output> apply(Iterable<Input> input) {
+
+        PythonWorkerManager<Input, Output> manager = new PythonWorkerManager<>(myUDF, serializedUDF, input);
+        Iterable<Output> output = manager.execute();
+        return output;
+    }
+
+}
diff --git a/wayang-api/wayang-api-python/src/main/python/worker.py b/wayang-api/wayang-api-python/src/main/python/worker.py
new file mode 100644
index 0000000..6e3d54a
--- /dev/null
+++ b/wayang-api/wayang-api-python/src/main/python/worker.py
@@ -0,0 +1,178 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import os
+import socket
+import struct
+import pickle
+import base64
+import re
+import sys
+
+
+class SpecialLengths(object):
+    END_OF_DATA_SECTION = -1
+    PYTHON_EXCEPTION_THROWN = -2
+    TIMING_DATA = -3
+    END_OF_STREAM = -4
+    NULL = -5
+    START_ARROW_STREAM = -6
+
+
+def read_int(stream):
+    length = stream.read(4)
+    if not length:
+        raise EOFError
+    res = struct.unpack("!i", length)[0]
+    return res
+
+
+class UTF8Deserializer:
+    """
+    Deserializes streams written by String.getBytes.
+    """
+
+    def __init__(self, use_unicode=True):
+        self.use_unicode = use_unicode
+
+    def loads(self, stream):
+        length = read_int(stream)
+        if length == SpecialLengths.END_OF_DATA_SECTION:
+            raise EOFError
+        elif length == SpecialLengths.NULL:
+            return None
+        s = stream.read(length)
+        return s.decode("utf-8") if self.use_unicode else s
+
+    def load_stream(self, stream):
+        try:
+            while True:
+                yield self.loads(stream)
+        except struct.error:
+            return
+        except EOFError:
+            return
+
+    def __repr__(self):
+        return "UTF8Deserializer(%s)" % self.use_unicode
+
+
+def write_int(p, outfile):
+    outfile.write(struct.pack("!i", p))
+
+
+def write_with_length(obj, stream):
+    serialized = obj.encode('utf-8')
+    if serialized is None:
+        raise ValueError("serialized value should not be None")
+    if len(serialized) > (1 << 31):
+        raise ValueError("can not serialize object larger than 2G")
+    write_int(len(serialized), stream)
+    stream.write(serialized)
+
+
+def dump_stream(iterator, stream):
+
+    for obj in iterator:
+        if type(obj) is str:
+            write_with_length(obj, stream)
+        ## elif type(obj) is list:
+        ##    write_with_length(obj, stream)
+    print("Termine")
+    write_int(SpecialLengths.END_OF_DATA_SECTION, stream)
+    print("Escribi Fin")
+
+
+def process(infile, outfile):
+    """udf64 = os.environ["UDF"]
+    print("udf64")
+    print(udf64)
+    #serialized_udf = binascii.a2b_base64(udf64)
+    #serialized_udf = base64.b64decode(udf64)
+    serialized_udf = bytearray(udf64, encoding='utf-16')
+    # NOT VALID TO BE UTF8  serialized_udf = bytes(udf64, 'UTF-8')
+    print("serialized_udf")
+    print(serialized_udf)
+    # input to be ast.literal_eval(serialized_udf)
+    func = pickle.loads(serialized_udf, encoding="bytes")
+    print ("func")
+    print (func)
+    print(func([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))
+    # func([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])"""
+
+
+
+    # TODO First we must receive the operator + UDF
+    """udf = lambda elem: elem.lower()
+
+    def func(it):
+        return sorted(it, key=udf)"""
+    udf_length = read_int(infile)
+    print("udf_length")
+    print(udf_length)
+    serialized_udf = infile.read(udf_length)
+    print("serialized_udf")
+    print(serialized_udf)
+    #base64_message = base64.b64decode(serialized_udf + "===")
+    #print("base64_message")
+    #print(base64_message)
+    func = pickle.loads(serialized_udf)
+    #func = ori.lala(serialized_udf)
+    #print (func)
+    #for x in func([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]): print(x)
+
+    """print("example")
+    for x in func("2344|234|efrf|$#|ffrf"): print(x)"""
+    # TODO Here we are temporarily assuming that the user is exclusively sending UTF8. User has several types
+    iterator = UTF8Deserializer().load_stream(infile)
+    # out_iter = sorted(iterator, key=lambda elem: elem.lower())
+    out_iter = func(iterator)
+    dump_stream(iterator=out_iter, stream=outfile)
+
+
+def local_connect(port):
+    sock = None
+    errors = []
+    # Support for both IPv4 and IPv6.
+    # On most of IPv6-ready systems, IPv6 will take precedence.
+    for res in socket.getaddrinfo("127.0.0.1", port, socket.AF_UNSPEC, socket.SOCK_STREAM):
+        af, socktype, proto, _, sa = res
+        try:
+            sock = socket.socket(af, socktype, proto)
+            # sock.settimeout(int(os.environ.get("SPARK_AUTH_SOCKET_TIMEOUT", 15)))
+            sock.settimeout(30)
+            sock.connect(sa)
+            # sockfile = sock.makefile("rwb", int(os.environ.get("SPARK_BUFFER_SIZE", 65536)))
+            sockfile = sock.makefile("rwb", 65536)
+            # _do_server_auth(sockfile, auth_secret)
+            return (sockfile, sock)
+        except socket.error as e:
+            emsg = str(e)
+            errors.append("tried to connect to %s, but an error occurred: %s" % (sa, emsg))
+            sock.close()
+            sock = None
+    raise Exception("could not open socket: %s" % errors)
+
+
+if __name__ == '__main__':
+    print("Python version")
+    print (sys.version)
+    java_port = int(os.environ["PYTHON_WORKER_FACTORY_PORT"])
+    sock_file, sock = local_connect(java_port)
+    process(sock_file, sock_file)
+    sock_file.flush()
+    exit()
diff --git a/wayang-api/wayang-api-python/src/main/resources/wayang-api-python-defaults.properties b/wayang-api/wayang-api-python/src/main/resources/wayang-api-python-defaults.properties
index ff3a99f..99b8470 100644
--- a/wayang-api/wayang-api-python/src/main/resources/wayang-api-python-defaults.properties
+++ b/wayang-api/wayang-api-python/src/main/resources/wayang-api-python-defaults.properties
@@ -15,4 +15,5 @@
 # limitations under the License.
 #
 
-# TODO: Add properties here
\ No newline at end of file
+wayang.api.python.worker = /Users/rodrigopardomeza/wayang/incubator-wayang/wayang-api/wayang-api-python/src/main/python/worker.py
+# wayang.api.python.worker = src/main/python/worker.py
diff --git a/wayang-api/wayang-api-python/src/test/java/org/apache/wayang/api/python/BasicPython.java b/wayang-api/wayang-api-python/src/test/java/org/apache/wayang/api/python/BasicPython.java
new file mode 100644
index 0000000..5108de2
--- /dev/null
+++ b/wayang-api/wayang-api-python/src/test/java/org/apache/wayang/api/python/BasicPython.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.python;
+
+import org.apache.wayang.api.python.function.WrappedPythonFunction;
+import org.apache.wayang.basic.operators.*;
+import org.apache.wayang.core.api.WayangContext;
+import org.apache.wayang.core.function.FlatMapDescriptor;
+import org.apache.wayang.core.function.MapPartitionsDescriptor;
+import org.apache.wayang.core.function.TransformationDescriptor;
+import org.apache.wayang.core.plan.wayangplan.WayangPlan;
+import org.apache.wayang.core.types.DataUnitType;
+import org.apache.wayang.java.Java;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Arrays;
+
+public class BasicPython {
+
+    public static URI createUri(String resourcePath) {
+        try {
+            return Thread.currentThread().getClass().getResource(resourcePath).toURI();
+        } catch (URISyntaxException e) {
+            throw new IllegalArgumentException("Illegal URI.", e);
+        }
+
+    }
+
+    @Test
+    public void testOnJava() throws URISyntaxException, IOException {
+
+        URI FILE_SOME_LINES_TXT = createUri("/python-lines.txt");
+
+        // Instantiate Rheem and activate the backend.
+        WayangContext rheemContext = new WayangContext().with(Java.basicPlugin());
+        TextFileSource textFileSource = new TextFileSource(FILE_SOME_LINES_TXT.toString());
+
+        TextFileSink<String> sink = new TextFileSink<String>(
+                "file:///Users/rodrigopardomeza/wayang/incubator-wayang/wayang-api/wayang-api-python/src/main/resources/basic_output.txt",
+                String.class
+        );
+
+        textFileSource.connectTo(0, sink, 0);
+
+        rheemContext.execute(new WayangPlan(sink));
+    }
+}
+
diff --git a/wayang-api/wayang-api-python/src/test/java/org/apache/wayang/api/python/SortPython.java b/wayang-api/wayang-api-python/src/test/java/org/apache/wayang/api/python/SortPython.java
new file mode 100644
index 0000000..7b74e77
--- /dev/null
+++ b/wayang-api/wayang-api-python/src/test/java/org/apache/wayang/api/python/SortPython.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.wayang.api.python;
+
+import org.apache.wayang.api.python.function.WrappedPythonFunction;
+import org.apache.wayang.basic.operators.*;
+import org.apache.wayang.core.api.WayangContext;
+import org.apache.wayang.core.function.FlatMapDescriptor;
+import org.apache.wayang.core.function.MapPartitionsDescriptor;
+import org.apache.wayang.core.function.TransformationDescriptor;
+import org.apache.wayang.core.plan.wayangplan.WayangPlan;
+import org.apache.wayang.core.types.DataUnitType;
+import org.apache.wayang.java.Java;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Arrays;
+
+public class SortPython {
+    /*
+    public static URI createUri(String resourcePath) {
+        try {
+            return Thread.currentThread().getClass().getResource(resourcePath).toURI();
+        } catch (URISyntaxException e) {
+            throw new IllegalArgumentException("Illegal URI.", e);
+        }
+
+    }
+
+    @Test
+    public void testOnJava() throws URISyntaxException, IOException {
+
+        URI FILE_SOME_LINES_TXT = createUri("/python-lines.txt");
+
+        // Instantiate Rheem and activate the backend.
+        WayangContext rheemContext = new WayangContext().with(Java.basicPlugin());
+        TextFileSource textFileSource = new TextFileSource(FILE_SOME_LINES_TXT.toString());
+
+        // for each line (input) output an iterator of the words
+        FlatMapOperator<String, String> flatMapOperator
+                = new FlatMapOperator<>(
+                new FlatMapDescriptor<>(
+                        line -> Arrays.asList(
+                                (String[]) line.split(" ")
+                        ),
+                        String.class,
+                        String.class
+                )
+        );
+
+        SortOperator<String, String> sortJava
+                = new SortOperator<String, String>(
+                new TransformationDescriptor<String, String>(
+                        (l) -> l.toLowerCase(),
+                        DataUnitType.createBasic(String.class),
+                        DataUnitType.createBasic(String.class)
+                )
+        );
+
+        MapPartitionsOperator<String, String> sortPython =
+                new MapPartitionsOperator<String, String>(
+                        new MapPartitionsDescriptor<String, String>(
+                                new WrappedPythonFunction<String, String>(
+                                        l -> l
+                                ),
+                                String.class,
+                                String.class
+                        )
+                );
+
+        //URI outputfile = Thread.currentThread().getClass().getResource("/output.txt").toURI();
+
+        TextFileSink<String> sink = new TextFileSink<String>(
+                "file:///Users/rodrigopardomeza/wayang/incubator-wayang/wayang-api/wayang-api-python/src/main/resources/output.txt",
+                String.class
+        );
+
+        textFileSource.connectTo(0, flatMapOperator, 0);
+        flatMapOperator.connectTo(0, sortPython, 0);
+        sortPython.connectTo(0, sink, 0);
+
+        rheemContext.execute(new WayangPlan(sink));
+    }
+     */
+}
diff --git a/wayang-api/wayang-api-rest/pom.xml b/wayang-api/wayang-api-rest/pom.xml
new file mode 100644
index 0000000..ef86974
--- /dev/null
+++ b/wayang-api/wayang-api-rest/pom.xml
@@ -0,0 +1,127 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing,
+  software distributed under the License is distributed on an
+  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  KIND, either express or implied.  See the License for the
+  specific language governing permissions and limitations
+  under the License.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>wayang-api</artifactId>
+        <groupId>org.apache.wayang</groupId>
+        <version>0.6.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>wayang-api-rest</artifactId>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-dependencies</artifactId>
+                <version>2.2.6.RELEASE</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+<!--        TODO add the version fo protobuf with parameter-->
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+            <version>3.15.2</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-core</artifactId>
+            <version>0.6.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-java</artifactId>
+            <version>0.6.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-api-python</artifactId>
+            <version>0.6.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-serializable</artifactId>
+            <version>0.6.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-api-python</artifactId>
+            <version>0.6.1-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+    <profiles>
+        <profile>
+            <id>java8</id>
+            <activation>
+                <jdk>1.8</jdk>
+            </activation>
+            <dependencies>
+                <dependency>
+                    <groupId>org.apache.wayang</groupId>
+                    <artifactId>wayang-spark_2.11</artifactId>
+                    <version>0.6.1-SNAPSHOT</version>
+                </dependency>
+            </dependencies>
+        </profile>
+        <profile>
+            <id>java11</id>
+            <activation>
+                <jdk>11</jdk>
+            </activation>
+            <dependencies>
+                <dependency>
+                    <groupId>org.apache.wayang</groupId>
+                    <artifactId>wayang-spark_2.12</artifactId>
+                    <version>0.6.1-SNAPSHOT</version>
+                </dependency>
+            </dependencies>
+        </profile>
+    </profiles>
+</project>
\ No newline at end of file
diff --git a/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/WayangRESTAPI.java b/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/WayangRESTAPI.java
new file mode 100644
index 0000000..8b37b7c
--- /dev/null
+++ b/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/WayangRESTAPI.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.rest;
+
+public class WayangRESTAPI {
+}
diff --git a/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/server/spring/WayangApplication.java b/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/server/spring/WayangApplication.java
new file mode 100644
index 0000000..b9b6a76
--- /dev/null
+++ b/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/server/spring/WayangApplication.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.rest.server.spring;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class WayangApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(WayangApplication.class, args);
+    }
+}
diff --git a/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/server/spring/decoder/WayangPlanBuilder.java b/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/server/spring/decoder/WayangPlanBuilder.java
new file mode 100644
index 0000000..649c5ef
--- /dev/null
+++ b/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/server/spring/decoder/WayangPlanBuilder.java
@@ -0,0 +1,240 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.rest.server.spring.decoder;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.wayang.api.python.function.WrappedPythonFunction;
+import org.apache.wayang.basic.operators.MapPartitionsOperator;
+import org.apache.wayang.basic.operators.TextFileSink;
+import org.apache.wayang.basic.operators.TextFileSource;
+import org.apache.wayang.basic.operators.UnionAllOperator;
+import org.apache.wayang.commons.serializable.OperatorProto;
+import org.apache.wayang.commons.serializable.PlanProto;
+import org.apache.wayang.commons.serializable.WayangPlanProto;
+import org.apache.wayang.core.api.WayangContext;
+import org.apache.wayang.core.api.exception.WayangException;
+import org.apache.wayang.core.function.MapPartitionsDescriptor;
+import org.apache.wayang.core.plan.wayangplan.OperatorBase;
+import org.apache.wayang.core.plan.wayangplan.WayangPlan;
+import org.apache.wayang.java.Java;
+import org.apache.wayang.spark.Spark;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.*;
+import java.util.stream.Collectors;
+import java.util.Base64;
+
+public class WayangPlanBuilder {
+
+    private WayangPlan wayangPlan;
+    private WayangContext wayangContext;
+
+    public WayangPlanBuilder(FileInputStream planFile){
+        try {
+
+            WayangPlanProto plan = WayangPlanProto.parseFrom(planFile);
+
+            this.wayangContext = buildContext(plan);
+            this.wayangPlan = buildPlan(plan);
+
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public WayangPlanBuilder(String writtenPlan){
+
+        System.out.println(writtenPlan);
+        byte[] message = Base64.getDecoder().decode(writtenPlan);
+        System.out.println(message);
+
+        try {
+            WayangPlanProto plan = WayangPlanProto.parseFrom(message);
+
+            this.wayangContext = buildContext(plan);
+            this.wayangPlan = buildPlan(plan);
+        } catch (InvalidProtocolBufferException e) {
+            e.printStackTrace();
+        }
+
+    }
+
+    private WayangContext buildContext(WayangPlanProto plan){
+
+        WayangContext ctx = new WayangContext();
+        plan.getContext().getPlatformsList().forEach(platform -> {
+            if (platform.getNumber() == 0)
+                ctx.with(Java.basicPlugin());
+            else if (platform.getNumber() == 1)
+                ctx.with(Spark.basicPlugin());
+        });
+        //ctx.with(Spark.basicPlugin());
+
+        return ctx;
+    }
+
+    private WayangPlan buildPlan(WayangPlanProto plan){
+
+        System.out.println(plan);
+
+        PlanProto planProto = plan.getPlan();
+        LinkedList<OperatorProto> protoList = new LinkedList<>();
+        planProto.getSourcesList().forEach(protoList::addLast);
+
+        Map<String, OperatorBase> operators = new HashMap<>();
+        List<OperatorBase> sinks = new ArrayList<>();
+        while(! protoList.isEmpty()) {
+
+            OperatorProto proto = protoList.pollFirst();
+
+            /* Checking if protoOperator can be connected to the current WayangPlan*/
+            boolean processIt;
+            if(proto.getType().equals("source")) processIt = true;
+
+            else {
+                /* Checking if ALL predecessors were already processed */
+                processIt = true;
+                for(String predecessor : proto.getPredecessorsList()){
+                    if (!operators.containsKey(predecessor)) {
+                        processIt = false;
+                        break;
+                    }
+                }
+            }
+
+            /* Operators should not be processed twice*/
+            if(operators.containsKey(proto.getId())) processIt = false;
+
+            if(processIt) {
+
+                /* Create and store Wayang operator */
+                OperatorBase operator = createOperatorByType(proto);
+                operators.put(proto.getId(), operator);
+
+                /*TODO Connect with predecessors requires more details in connection slot*/
+                int order = 0;
+                for (String pre_id : proto.getPredecessorsList()) {
+
+                    OperatorBase predecessor = operators.get(pre_id);
+                    /* Only works without replicate topology */
+                    predecessor.connectTo(0, operator, order);
+                    order++;
+
+                    if(proto.getType().equals("sink")){
+                        sinks.add(operator);
+                        //if(!sinks.contains(operator)) {
+                        //    sinks.add(operator);
+                        //}
+                    }
+                }
+
+                /*List of OperatorProto successors
+                 * They will be added to the protoList
+                 * nevertheless they must be processed only if the parents are in operators list */
+                List<OperatorProto> listSuccessors = planProto.getOperatorsList()
+                        .stream()
+                        .filter(t -> proto.getSuccessorsList().contains(t.getId()))
+                        .collect(Collectors.toList());
+                for (OperatorProto successor : listSuccessors){
+                    if(!protoList.contains(successor)){
+                        protoList.addLast(successor);
+                    }
+                }
+
+                List<OperatorProto> sinkSuccessors = planProto.getSinksList()
+                        .stream()
+                        .filter(t -> proto.getSuccessorsList().contains(t.getId()))
+                        .collect(Collectors.toList());
+                for (OperatorProto successor : sinkSuccessors){
+                    if(!protoList.contains(successor)){
+                        protoList.addLast(successor);
+                    }
+                }
+
+            } else {
+
+                /* In case we cannot process it yet, It must be added again at the end*/
+                protoList.addLast(proto);
+            }
+        }
+
+        WayangPlan wayangPlan = new WayangPlan(sinks.get(0));
+        return wayangPlan;
+    }
+
+    public OperatorBase createOperatorByType(OperatorProto operator){
+
+        System.out.println("Typo: " + operator.getType());
+        switch(operator.getType()){
+            case "source":
+                try {
+                    String source_path = operator.getPath();
+                    URL url = new File(source_path).toURI().toURL();
+                    return new TextFileSource(url.toString());
+                } catch (MalformedURLException e) {
+                    e.printStackTrace();
+                }
+                break;
+            case "sink":
+                try {
+                    String sink_path = operator.getPath();
+                    URL url = new File(sink_path).toURI().toURL();
+                    return new TextFileSink<String>(
+                            url.toString(),
+                            String.class
+                    );
+
+                } catch (MalformedURLException e) {
+                    e.printStackTrace();
+                }
+                break;
+            case "map_partition":
+                return new MapPartitionsOperator<>(
+                        new MapPartitionsDescriptor<String, String>(
+                                new WrappedPythonFunction<String, String>(
+                                        l -> l,
+                                        operator.getUdf()
+                                ),
+                                String.class,
+                                String.class
+                        )
+                );
+
+            case "union":
+                return new UnionAllOperator<String>(
+                        String.class
+                );
+
+        }
+
+        throw new WayangException("Operator Type not supported");
+    }
+
+    public WayangContext getWayangContext() {
+        return wayangContext;
+    }
+
+    public WayangPlan getWayangPlan() {
+        return wayangPlan;
+    }
+}
diff --git a/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/server/spring/general/WayangController.java b/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/server/spring/general/WayangController.java
new file mode 100644
index 0000000..4f48863
--- /dev/null
+++ b/wayang-api/wayang-api-rest/src/main/java/org/apache/wayang/api/rest/server/spring/general/WayangController.java
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.api.rest.server.spring.general;
+
+import com.google.protobuf.ByteString;
+import org.apache.wayang.api.python.function.WrappedPythonFunction;
+import org.apache.wayang.api.rest.server.spring.decoder.WayangPlanBuilder;
+import org.apache.wayang.basic.operators.*;
+import org.apache.wayang.commons.serializable.OperatorProto;
+import org.apache.wayang.commons.serializable.PlanProto;
+import org.apache.wayang.core.api.WayangContext;
+import org.apache.wayang.core.api.exception.WayangException;
+import org.apache.wayang.core.function.MapPartitionsDescriptor;
+import org.apache.wayang.core.plan.wayangplan.OperatorBase;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Paths;
+import java.util.*;
+import java.util.stream.Collectors;
+
+import org.apache.wayang.core.plan.wayangplan.WayangPlan;
+import org.apache.wayang.java.Java;
+import org.apache.wayang.spark.Spark;
+
+import org.apache.wayang.commons.serializable.WayangPlanProto;
+import org.springframework.web.multipart.MultipartFile;
+
+
+@RestController
+public class WayangController {
+
+    @GetMapping("/plan/create/fromfile")
+    public String planFromFile(
+            //@RequestParam("file") MultipartFile file
+    ){
+
+        try {
+            FileInputStream inputStream = new FileInputStream(Paths.get(".").toRealPath() + "/protobuf/wayang_message");
+            WayangPlanBuilder wpb = new WayangPlanBuilder(inputStream);
+
+            /*TODO ADD id to executions*/
+            wpb.getWayangContext().execute(wpb.getWayangPlan());
+
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+
+        return "Builder works";
+    }
+
+    @PostMapping("/plan/create")
+    public String planFromMessage(
+            @RequestParam("message") String message
+    ){
+
+        WayangPlanBuilder wpb = new WayangPlanBuilder(message);
+
+        /*TODO ADD id to executions*/
+        wpb.getWayangContext().execute(wpb.getWayangPlan());
+
+        return "";
+    }
+
+    @GetMapping("/")
+    public String all(){
+        System.out.println("detected!");
+
+        try {
+            FileInputStream inputStream = new FileInputStream(Paths.get(".").toRealPath() + "/protobuf/wayang_message");
+            WayangPlanProto plan = WayangPlanProto.parseFrom(inputStream);
+
+            WayangContext wc = buildContext(plan);
+            WayangPlan wp = buildPlan(plan);
+
+            System.out.println("Plan!");
+            System.out.println(wp.toString());
+
+            wc.execute(wp);
+            return("Works!");
+
+        } catch (FileNotFoundException e) {
+            e.printStackTrace();
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+
+        return "Not working";
+    }
+
+    private WayangContext buildContext(WayangPlanProto plan){
+
+        WayangContext ctx = new WayangContext();
+        plan.getContext().getPlatformsList().forEach(platform -> {
+            if (platform.getNumber() == 0)
+                ctx.with(Java.basicPlugin());
+            else if (platform.getNumber() == 1)
+                ctx.with(Spark.basicPlugin());
+        });
+        //ctx.with(Spark.basicPlugin());
+
+        return ctx;
+    }
+
+    private WayangPlan buildPlan(WayangPlanProto plan){
+
+        System.out.println(plan);
+
+        PlanProto planProto = plan.getPlan();
+        LinkedList<OperatorProto> protoList = new LinkedList<>();
+        planProto.getSourcesList().forEach(protoList::addLast);
+
+        Map<String, OperatorBase> operators = new HashMap<>();
+        List<OperatorBase> sinks = new ArrayList<>();
+        while(! protoList.isEmpty()) {
+
+            OperatorProto proto = protoList.pollFirst();
+
+            /* Checking if protoOperator can be connected to the current WayangPlan*/
+            boolean processIt;
+            if(proto.getType().equals("source")) processIt = true;
+
+            else {
+                /* Checking if ALL predecessors were already processed */
+                processIt = true;
+                for(String predecessor : proto.getPredecessorsList()){
+                    if (!operators.containsKey(predecessor)) {
+                        processIt = false;
+                        break;
+                    }
+                }
+            }
+
+            /* Operators should not be processed twice*/
+            if(operators.containsKey(proto.getId())) processIt = false;
+
+            if(processIt) {
+
+                /* Create and store Wayang operator */
+                OperatorBase operator = createOperatorByType(proto);
+                operators.put(proto.getId(), operator);
+
+                /*TODO Connect with predecessors requires more details in connection slot*/
+                int order = 0;
+                for (String pre_id : proto.getPredecessorsList()) {
+
+                    OperatorBase predecessor = operators.get(pre_id);
+                    /* Only works without replicate topology */
+                    predecessor.connectTo(0, operator, order);
+                    order++;
+
+                    if(proto.getType().equals("sink")){
+                        sinks.add(operator);
+                        //if(!sinks.contains(operator)) {
+                        //    sinks.add(operator);
+                        //}
+                    }
+                }
+
+                /*List of OperatorProto successors
+                 * They will be added to the protoList
+                 * nevertheless they must be processed only if the parents are in operators list */
+                List<OperatorProto> listSuccessors = planProto.getOperatorsList()
+                        .stream()
+                        .filter(t -> proto.getSuccessorsList().contains(t.getId()))
+                        .collect(Collectors.toList());
+                for (OperatorProto successor : listSuccessors){
+                    if(!protoList.contains(successor)){
+                        protoList.addLast(successor);
+                    }
+                }
+
+                List<OperatorProto> sinkSuccessors = planProto.getSinksList()
+                        .stream()
+                        .filter(t -> proto.getSuccessorsList().contains(t.getId()))
+                        .collect(Collectors.toList());
+                for (OperatorProto successor : sinkSuccessors){
+                    if(!protoList.contains(successor)){
+                        protoList.addLast(successor);
+                    }
+                }
+
+            } else {
+
+                /* In case we cannot process it yet, It must be added again at the end*/
+                protoList.addLast(proto);
+            }
+        }
+
+        WayangPlan wayangPlan = new WayangPlan(sinks.get(0));
+        return wayangPlan;
+    }
+
+    public OperatorBase createOperatorByType(OperatorProto operator){
+
+        System.out.println("Typo: " + operator.getType());
+        switch(operator.getType()){
+            case "source":
+                try {
+                    String source_path = operator.getPath();
+                    URL url = new File(source_path).toURI().toURL();
+                    return new TextFileSource(url.toString());
+                } catch (MalformedURLException e) {
+                    e.printStackTrace();
+                }
+                break;
+            case "sink":
+                try {
+                    String sink_path = operator.getPath();
+                    URL url = new File(sink_path).toURI().toURL();
+                    return new TextFileSink<String>(
+                            url.toString(),
+                            String.class
+                    );
+
+                } catch (MalformedURLException e) {
+                    e.printStackTrace();
+                }
+                break;
+            case "reduce_by_key":
+                try {
+                    /* Function to be applied in Python workers */
+                    ByteString function = operator.getUdf();
+
+                    /* Has dimension or positions that compose GroupKey */
+                    Map<String, String> parameters = operator.getParametersMap();
+
+                    PyWayangReduceByOperator<String, String> op = new PyWayangReduceByOperator(
+                        operator.getParametersMap(),
+                        operator.getUdf() ,
+                        String.class,
+                        String.class,
+                            false
+                    );
+
+                    String sink_path = operator.getPath();
+                    URL url = new File(sink_path).toURI().toURL();
+                    return new TextFileSink<String>(
+                            url.toString(),
+                            String.class
+                    );
+
+                } catch (MalformedURLException e) {
+                    e.printStackTrace();
+                }
+                break;
+            case "map_partition":
+                return new MapPartitionsOperator<>(
+                    new MapPartitionsDescriptor<String, String>(
+                        new WrappedPythonFunction<String, String>(
+                            l -> l,
+                            operator.getUdf()
+                        ),
+                        String.class,
+                        String.class
+                    )
+                );
+
+            case "union":
+                return new UnionAllOperator<String>(
+                        String.class
+                );
+
+        }
+
+        throw new WayangException("Operator Type not supported");
+    }
+
+    public static URI createUri(String resourcePath) {
+        try {
+            return Thread.currentThread().getClass().getResource(resourcePath).toURI();
+        } catch (URISyntaxException e) {
+            throw new IllegalArgumentException("Illegal URI.", e);
+        }
+
+    }
+
+}
diff --git a/wayang-api/wayang-api-rest/src/main/resources/application.properties b/wayang-api/wayang-api-rest/src/main/resources/application.properties
new file mode 100644
index 0000000..14b1a83
--- /dev/null
+++ b/wayang-api/wayang-api-rest/src/main/resources/application.properties
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+server.port = 8080
diff --git a/wayang-commons/pom.xml b/wayang-commons/pom.xml
index a0f1d18..58db211 100644
--- a/wayang-commons/pom.xml
+++ b/wayang-commons/pom.xml
@@ -209,6 +209,7 @@
         <module>wayang-core</module>
         <module>wayang-basic</module>
         <module>wayang-utils-profile-db</module>
+        <module>wayang-serializable</module>
     </modules>
 
 </project>
diff --git a/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/PyWayangReduceByOperator.java b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/PyWayangReduceByOperator.java
new file mode 100644
index 0000000..1951b31
--- /dev/null
+++ b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/PyWayangReduceByOperator.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.basic.operators;
+
+import com.google.protobuf.ByteString;
+import org.apache.wayang.core.plan.wayangplan.UnaryToUnaryOperator;
+import org.apache.wayang.core.types.DataSetType;
+
+import java.util.Map;
+
+public class PyWayangReduceByOperator<Type, Key> extends UnaryToUnaryOperator<Type, Type> {
+
+    protected final Map<String, String> parameters;
+    protected final ByteString reduceDescriptor;
+
+    public PyWayangReduceByOperator(Map<String, String> parameters,
+                                       ByteString reduceDescriptor,
+                                       DataSetType<Type> inputType, DataSetType<Type> outputType, boolean isSupportingBroadcastInputs) {
+        super(inputType, outputType, isSupportingBroadcastInputs);
+        this.parameters = parameters;
+        this.reduceDescriptor = reduceDescriptor;
+    }
+
+    public PyWayangReduceByOperator(Map<String, String> parameters,
+                                    ByteString reduceDescriptor,
+                                    Class<Type> inputType, Class<Type> outputType, boolean isSupportingBroadcastInputs) {
+        super(DataSetType.createDefault(inputType), DataSetType.createDefault(outputType), isSupportingBroadcastInputs);
+        this.parameters = parameters;
+        this.reduceDescriptor = reduceDescriptor;
+    }
+
+    public PyWayangReduceByOperator(PyWayangReduceByOperator<Type, Type> that) {
+        super(that);
+        this.parameters = that.getParameters();
+        this.reduceDescriptor = that.getReduceDescriptor();
+    }
+
+    public Map<String, String>  getParameters() {
+        return this.parameters;
+    }
+
+    public ByteString getReduceDescriptor() {
+        return this.reduceDescriptor;
+    }
+
+}
\ No newline at end of file
diff --git a/wayang-commons/wayang-serializable/pom.xml b/wayang-commons/wayang-serializable/pom.xml
new file mode 100644
index 0000000..87b17ea
--- /dev/null
+++ b/wayang-commons/wayang-serializable/pom.xml
@@ -0,0 +1,131 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing,
+  software distributed under the License is distributed on an
+  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  KIND, either express or implied.  See the License for the
+  specific language governing permissions and limitations
+  under the License.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>wayang-commons</artifactId>
+        <groupId>org.apache.wayang</groupId>
+        <version>0.6.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>wayang-serializable</artifactId>
+    <packaging>jar</packaging>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <protobuf.version>3.15.2</protobuf.version>
+        <protoc.version>3.15.2</protoc.version>
+        <maven.compiler.source>1.8</maven.compiler.source>
+        <maven.compiler.target>1.8</maven.compiler.target>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+            <version>${protobuf.version}</version>
+        </dependency>
+    </dependencies>
+    <build>
+        <extensions>
+            <extension>
+                <groupId>kr.motd.maven</groupId>
+                <artifactId>os-maven-plugin</artifactId>
+                <version>1.6.2</version>
+            </extension>
+        </extensions>
+        <plugins>
+            <plugin>
+                <groupId>org.xolstice.maven.plugins</groupId>
+                <artifactId>protobuf-maven-plugin</artifactId>
+                <version>0.5.1</version>
+                <configuration>
+                    <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>compile</goal>
+                            <goal>compile-python</goal>
+                            <goal>test-compile</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+        <!-- Copy all the protobuf code to the folder PyWayang -->
+        <resources>
+            <resource>
+                <!-- TODO: watch if is it possible to put it with parameters -->
+                <directory>${basedir}/target/generated-sources/protobuf/python</directory><!-- from -->
+                <targetPath>${basedir}/../../pywayang/protobuf</targetPath><!-- to -->
+                <includes><!-- Only python files -->
+                    <include>**/*.py</include>
+                </includes>
+            </resource>
+        </resources>
+    </build>
+
+    <profiles>
+        <profile>
+            <id>protof-unix</id>
+            <activation>
+                <os>
+                    <family>unix</family>
+                </os>
+            </activation>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.xolstice.maven.plugins</groupId>
+                        <artifactId>protobuf-maven-plugin</artifactId>
+                        <configuration>
+                            <!--This one is for Linux installation -->
+                            <protocExecutable>/usr/bin/protoc</protocExecutable>
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+        <profile>
+            <id>protof-mac</id>
+            <activation>
+                <os>
+                    <family>mac</family>
+                </os>
+            </activation>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.xolstice.maven.plugins</groupId>
+                        <artifactId>protobuf-maven-plugin</artifactId>
+                        <configuration>
+                            <!--This one is for Mac executions-->
+                            <protocExecutable>/usr/local/bin/protoc</protocExecutable>
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
+
+</project>
\ No newline at end of file
diff --git a/wayang-commons/wayang-serializable/src/main/proto/pywayangplan.proto b/wayang-commons/wayang-serializable/src/main/proto/pywayangplan.proto
new file mode 100644
index 0000000..5fb373e
--- /dev/null
+++ b/wayang-commons/wayang-serializable/src/main/proto/pywayangplan.proto
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax="proto2";
+
+option java_multiple_files = true;
+option java_package = "org.apache.wayang.commons.serializable";
+
+/*Java Operator or concatenated Python Pipeline*/
+message OperatorProto{
+  optional string id = 1;
+  optional string type = 2;
+  optional bytes udf = 3;
+  optional string path = 4;
+  // repeated OperatorProto predecessors = 5;
+  repeated string predecessors = 5;
+  // repeated OperatorProto successors = 6;
+  repeated string successors = 6;
+  map<string, string> parameters = 7;
+}
+
+message PlanProto{
+
+  enum TypesProto{
+    string = 0;
+    integer = 1;
+  }
+
+  repeated OperatorProto sources = 1;
+  repeated OperatorProto operators = 2;
+  repeated OperatorProto sinks = 3;
+  optional TypesProto input = 4;
+  optional TypesProto output = 5;
+
+}
+
+message ContextProto{
+
+  enum PlatformProto{
+    java = 0;
+    spark = 1;
+  }
+
+  repeated PlatformProto platforms = 1;
+}
+
+message WayangPlanProto{
+
+  optional PlanProto plan = 1;
+  optional ContextProto context = 2;
+
+}
\ No newline at end of file
diff --git a/wayang-commons/wayang-serializable/src/main/python/pyplangenerator.sh b/wayang-commons/wayang-serializable/src/main/python/pyplangenerator.sh
new file mode 100644
index 0000000..f86efb5
--- /dev/null
+++ b/wayang-commons/wayang-serializable/src/main/python/pyplangenerator.sh
@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+protoc -I=/Users/rodrigopardomeza/wayang/incubator-wayang/protobuf \
+--python_out=/Users/rodrigopardomeza/wayang/incubator-wayang/pywayang/protobuf/ \
+/Users/rodrigopardomeza/wayang/incubator-wayang/protobuf/pywayangplan.proto
\ No newline at end of file