diff --git a/python/src/pywy/platforms/commons/channels.py b/python/src/pywy/platforms/commons/channels.py
index 743a716..57cebeb 100644
--- a/python/src/pywy/platforms/commons/channels.py
+++ b/python/src/pywy/platforms/commons/channels.py
@@ -17,6 +17,7 @@
 
 from typing import Callable
 from pywy.core import (Channel, ChannelDescriptor)
+from pywy.exception import PywyException
 
 
 class CommonsCallableChannel(Channel):
@@ -35,6 +36,11 @@
 
     @staticmethod
     def concatenate(function_a: Callable, function_b: Callable):
+        if function_a is None:
+            raise PywyException("the function_a can't be None")
+        if function_b is None:
+            return function_a
+
         def executable(iterable):
             return function_a(function_b(iterable))
         return executable
diff --git a/python/src/pywy/platforms/jvm/__init__.py b/python/src/pywy/platforms/jvm/__init__.py
new file mode 100644
index 0000000..d9e26de
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/__init__.py
@@ -0,0 +1,16 @@
+#
+#  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.
+#
diff --git a/python/src/pywy/platforms/jvm/channels.py b/python/src/pywy/platforms/jvm/channels.py
new file mode 100644
index 0000000..9c1e76f
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/channels.py
@@ -0,0 +1,47 @@
+#
+#  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 typing import Callable
+
+from pywy.core import (Channel, ChannelDescriptor)
+from pywy.exception import PywyException
+from pywy.platforms.commons.channels import CommonsCallableChannel
+from pywy.platforms.jvm.serializable.wayang_jvm_operator import WayangJVMOperator, WayangJVMMappartitionOperator
+
+
+class DispatchableChannel(CommonsCallableChannel):
+
+    operator: WayangJVMOperator
+
+    def __init__(self):
+        Channel.__init__(self)
+        self.udf = None
+        self.operator = None
+
+    def provide_dispatchable(self, do_wrapper: bool = False) -> WayangJVMOperator:
+        if self.operator is None:
+            raise PywyException("The operator was not define")
+        if do_wrapper:
+            self.operator.udf = self.udf
+
+        return self.operator
+
+    def accept_dispatchable(self, operator: WayangJVMOperator) -> 'WayangJVMOperator':
+        self.operator = operator
+        return self
+
+
+DISPATCHABLE_CHANNEL_DESCRIPTOR = ChannelDescriptor(type(DispatchableChannel()), True, True)
diff --git a/python/src/pywy/platforms/jvm/execution.py b/python/src/pywy/platforms/jvm/execution.py
new file mode 100644
index 0000000..2dad76d
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/execution.py
@@ -0,0 +1,99 @@
+#
+#  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 pywy.core import Executor, ChannelDescriptor
+from pywy.core import PywyPlan
+from pywy.platforms.jvm.channels import DISPATCHABLE_CHANNEL_DESCRIPTOR
+from pywy.platforms.jvm.graph import NodeDispatch, WGraphDispatch
+from pywy.platforms.jvm.operator import JVMExecutionOperator
+
+class JVMExecutor(Executor):
+
+    def __init__(self):
+        super(JVMExecutor, self).__init__()
+
+    def execute(self, plan):
+        pywyPlan: PywyPlan = plan
+        graph = WGraphDispatch(pywyPlan.sinks)
+
+        # TODO get this information by a configuration and ideally by the context
+        descriptor_default: ChannelDescriptor = DISPATCHABLE_CHANNEL_DESCRIPTOR
+
+        def execute(op_current: NodeDispatch, op_next: NodeDispatch):
+            if op_current is None:
+                return
+
+            jvm_current: JVMExecutionOperator = op_current.current
+            if jvm_current.outputs == 0:
+                jvm_current.execute(jvm_current.inputChannel, [])
+                return
+
+            if op_next is None:
+                return
+
+            jvm_next: JVMExecutionOperator = op_next.current
+            outputs = jvm_current.get_output_channeldescriptors()
+            inputs = jvm_next.get_input_channeldescriptors()
+
+            intersect = outputs.intersection(inputs)
+            if len(intersect) == 0:
+                raise Exception(
+                    "The operator(A) {} can't connect with (B) {}, "
+                    "because the output of (A) is {} and the input of (B) is {} ".format(
+                        jvm_current,
+                        jvm_next,
+                        outputs,
+                        inputs
+                    )
+                )
+
+            if len(intersect) > 1:
+                if descriptor_default is None:
+                    raise Exception(
+                        "The interaction between the operator (A) {} and (B) {}, "
+                        "can't be decided because are several channel availables {}".format(
+                            jvm_current,
+                            jvm_next,
+                            intersect
+                        )
+                    )
+                descriptor = descriptor_default
+            else:
+                descriptor = intersect.pop()
+
+            # TODO validate if is valite for several output
+            jvm_current.outputChannel[0] = descriptor.create_instance()
+
+            jvm_current.execute(jvm_current.inputChannel, jvm_current.outputChannel)
+
+            jvm_next.inputChannel = jvm_current.outputChannel
+
+        graph.traversal(graph.starting_nodes, execute)
+
+
+
+        starting: WayangJVMOperator = graph.starting_nodes[0].current.dispatch_operator
+        while starting.previous[0]:
+            print(starting)
+            #print(starting.nexts[0])
+            starting = starting.previous[0]
+            if len(starting.previous) == 0 :
+                break
+        print(starting)
+
+
+
diff --git a/python/src/pywy/platforms/jvm/graph.py b/python/src/pywy/platforms/jvm/graph.py
new file mode 100644
index 0000000..7863651
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/graph.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.
+#
+from typing import Iterable, List, Tuple
+
+from pywy.graph.graph import WayangGraph, GraphNode
+from pywy.operators.base import PO_T
+from pywy.platforms.jvm.serializable.wayang_jvm_operator import WJO_T
+
+
+class NodeDispatch(GraphNode[PO_T, PO_T]):
+    wop: WJO_T
+
+    def __init__(self, op: PO_T):
+        super(NodeDispatch, self).__init__(op)
+
+    def get_adjacents(self) -> List[PO_T]:
+        operator: PO_T = self.current
+        if operator is None or operator.inputs == 0:
+            return []
+        return operator.inputOperator
+
+    def build_node(self, t: PO_T) -> 'NodeDispatch':
+        return NodeDispatch(t)
+
+    def __str__(self):
+        return "NodeDispatch {}".format(self.current)
+
+    def __repr__(self):
+        return self.__str__()
+
+
+class WGraphDispatch(WayangGraph[PO_T, NodeDispatch]):
+
+    def __init__(self, nodes: Iterable[PO_T]):
+        super(WGraphDispatch, self).__init__(nodes)
+
+    def build_node(self, t: PO_T) -> NodeDispatch:
+        return NodeDispatch(t)
diff --git a/python/src/pywy/platforms/jvm/mappings.py b/python/src/pywy/platforms/jvm/mappings.py
new file mode 100644
index 0000000..4709a73
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/mappings.py
@@ -0,0 +1,26 @@
+#
+#  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 pywy.core import Mapping
+from pywy.platforms.jvm.operator import *
+
+
+JVM_OPERATOR_MAPPINGS = Mapping()
+
+JVM_OPERATOR_MAPPINGS.add_mapping(JVMFilterOperator())
+JVM_OPERATOR_MAPPINGS.add_mapping(JVMTextFileSourceOperator())
+JVM_OPERATOR_MAPPINGS.add_mapping(JVMTextFileSinkOperator())
diff --git a/python/src/pywy/platforms/jvm/operator/__init__.py b/python/src/pywy/platforms/jvm/operator/__init__.py
new file mode 100644
index 0000000..a0bc5d2
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/operator/__init__.py
@@ -0,0 +1,28 @@
+#
+#  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 pywy.platforms.jvm.operator.jvm_execution_operator import JVMExecutionOperator
+from pywy.platforms.jvm.operator.jvm_unary_filter import JVMFilterOperator
+from pywy.platforms.jvm.operator.jvm_source_textfile import JVMTextFileSourceOperator
+from pywy.platforms.jvm.operator.jvm_sink_textfile import JVMTextFileSinkOperator
+
+__ALL__ = [
+    JVMExecutionOperator,
+    JVMFilterOperator,
+    JVMTextFileSourceOperator,
+    JVMTextFileSinkOperator,
+]
diff --git a/python/src/pywy/platforms/jvm/operator/jvm_execution_operator.py b/python/src/pywy/platforms/jvm/operator/jvm_execution_operator.py
new file mode 100644
index 0000000..0049c5c
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/operator/jvm_execution_operator.py
@@ -0,0 +1,33 @@
+#
+#  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 typing import List, Type
+
+from pywy.core.channel import CH_T
+from pywy.operators.base import PywyOperator
+from pywy.platforms.jvm.serializable.wayang_jvm_operator import WayangJVMOperator
+
+
+class JVMExecutionOperator(PywyOperator):
+
+    dispatch_operator: WayangJVMOperator
+
+    def prefix(self) -> str:
+        return 'JVM'
+
+    def execute(self, inputs: List[Type[CH_T]], output: List[CH_T]):
+        pass
diff --git a/python/src/pywy/platforms/jvm/operator/jvm_sink_textfile.py b/python/src/pywy/platforms/jvm/operator/jvm_sink_textfile.py
new file mode 100644
index 0000000..555ba04
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/operator/jvm_sink_textfile.py
@@ -0,0 +1,56 @@
+#
+#  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 typing import Set, List, Type
+
+from pywy.core.channel import (CH_T, ChannelDescriptor)
+from pywy.exception import PywyException
+from pywy.operators.sink import TextFileSink
+from pywy.platforms.jvm.channels import DISPATCHABLE_CHANNEL_DESCRIPTOR, DispatchableChannel
+from pywy.platforms.jvm.operator.jvm_execution_operator import JVMExecutionOperator
+from pywy.platforms.jvm.serializable.wayang_jvm_operator import WayangJVMTextFileSink
+
+
+class JVMTextFileSinkOperator(TextFileSink, JVMExecutionOperator):
+
+    def __init__(self, origin: TextFileSink = None):
+        path = None if origin is None else origin.path
+        type_class = None if origin is None else origin.inputSlot[0]
+        end_line = None if origin is None else origin.end_line
+        super().__init__(path, type_class, end_line)
+
+    def execute(self, inputs: List[Type[CH_T]], outputs: List[Type[CH_T]]):
+        self.validate_channels(inputs, outputs)
+
+        if isinstance(inputs[0], DispatchableChannel):
+
+            py_in_dispatch_channel: DispatchableChannel = inputs[0]
+            operator = py_in_dispatch_channel.provide_dispatchable(do_wrapper=True)
+
+            sink: WayangJVMTextFileSink = WayangJVMTextFileSink(self.name, self.path)
+
+            operator.connect_to(0, sink, 0)
+
+            self.dispatch_operator = sink
+        else:
+            raise Exception("Channel Type does not supported")
+
+    def get_input_channeldescriptors(self) -> Set[ChannelDescriptor]:
+        return {DISPATCHABLE_CHANNEL_DESCRIPTOR}
+
+    def get_output_channeldescriptors(self) -> Set[ChannelDescriptor]:
+        raise Exception("The JVMTextFileSink does not support Output Channels")
diff --git a/python/src/pywy/platforms/jvm/operator/jvm_source_textfile.py b/python/src/pywy/platforms/jvm/operator/jvm_source_textfile.py
new file mode 100644
index 0000000..3ca3b91
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/operator/jvm_source_textfile.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.
+#
+
+from typing import Set, List, Type
+
+from pywy.core.channel import (CH_T, ChannelDescriptor)
+from pywy.operators.source import TextFileSource
+from pywy.platforms.jvm.channels import DISPATCHABLE_CHANNEL_DESCRIPTOR, DispatchableChannel
+from pywy.platforms.jvm.operator.jvm_execution_operator import JVMExecutionOperator
+from pywy.platforms.jvm.serializable.wayang_jvm_operator import WayangJVMTextFileSource
+
+
+class JVMTextFileSourceOperator(TextFileSource, JVMExecutionOperator):
+
+    def __init__(self, origin: TextFileSource = None):
+        path = None if origin is None else origin.path
+        super().__init__(path)
+        pass
+
+    def execute(self, inputs: List[Type[CH_T]], outputs: List[Type[CH_T]]):
+        self.validate_channels(inputs, outputs)
+        if isinstance(outputs[0], DispatchableChannel):
+            py_out_dispatch_channel: DispatchableChannel = outputs[0]
+            py_out_dispatch_channel.accept_dispatchable(
+                WayangJVMTextFileSource(self.name, self.path)
+            )
+        else:
+            raise Exception("Channel Type does not supported")
+
+    def get_input_channeldescriptors(self) -> Set[ChannelDescriptor]:
+        raise Exception("The JVMTextFileSource does not support Input Channels")
+
+    def get_output_channeldescriptors(self) -> Set[ChannelDescriptor]:
+        return {DISPATCHABLE_CHANNEL_DESCRIPTOR}
diff --git a/python/src/pywy/platforms/jvm/operator/jvm_unary_filter.py b/python/src/pywy/platforms/jvm/operator/jvm_unary_filter.py
new file mode 100644
index 0000000..06ba9e1
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/operator/jvm_unary_filter.py
@@ -0,0 +1,72 @@
+#
+#  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 typing import Set, List, Type
+
+from pywy.core.channel import CH_T, ChannelDescriptor
+from pywy.operators.unary import FilterOperator
+from pywy.platforms.jvm.channels import DISPATCHABLE_CHANNEL_DESCRIPTOR, DispatchableChannel
+from pywy.platforms.jvm.operator.jvm_execution_operator import JVMExecutionOperator
+from pywy.platforms.commons.channels import (
+    CommonsCallableChannel
+)
+from pywy.platforms.jvm.serializable.wayang_jvm_operator import WayangJVMMappartitionOperator, WayangJVMOperator
+
+
+class JVMFilterOperator(FilterOperator, JVMExecutionOperator):
+
+    def __init__(self, origin: FilterOperator = None):
+        predicate = None if origin is None else origin.predicate
+        super().__init__(predicate)
+        pass
+
+    def execute(self, inputs: List[Type[CH_T]], outputs: List[Type[CH_T]]):
+        self.validate_channels(inputs, outputs)
+        udf = self.predicate
+        if isinstance(inputs[0], DispatchableChannel):
+            py_in_dispatch_channel: DispatchableChannel = inputs[0]
+            py_out_dispatch_channel: DispatchableChannel = outputs[0]
+
+            def func(iterator):
+                return filter(udf, iterator)
+
+            py_out_dispatch_channel.accept_callable(
+                CommonsCallableChannel.concatenate(
+                    func,
+                    py_in_dispatch_channel.provide_callable()
+                )
+            )
+
+            op: WayangJVMOperator = py_in_dispatch_channel.provide_dispatchable()
+
+            if isinstance(op, WayangJVMMappartitionOperator):
+                py_out_dispatch_channel.accept_dispatchable(op)
+                return
+
+            current: WayangJVMMappartitionOperator = WayangJVMMappartitionOperator(self.name)
+            # TODO check for the case where the index matter
+            op.connect_to(0, current, 0)
+            py_out_dispatch_channel.accept_dispatchable(current)
+
+        else:
+            raise Exception("Channel Type does not supported")
+
+    def get_input_channeldescriptors(self) -> Set[ChannelDescriptor]:
+        return {DISPATCHABLE_CHANNEL_DESCRIPTOR}
+
+    def get_output_channeldescriptors(self) -> Set[ChannelDescriptor]:
+        return {DISPATCHABLE_CHANNEL_DESCRIPTOR}
diff --git a/python/src/pywy/platforms/jvm/platform.py b/python/src/pywy/platforms/jvm/platform.py
new file mode 100644
index 0000000..d52ac47
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/platform.py
@@ -0,0 +1,24 @@
+#
+#  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 pywy.core.platform import Platform
+
+
+class JVMPlatform(Platform):
+
+    def __init__(self):
+        super(JVMPlatform, self).__init__("JVM")
diff --git a/python/src/pywy/platforms/jvm/plugin.py b/python/src/pywy/platforms/jvm/plugin.py
new file mode 100644
index 0000000..eff1e04
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/plugin.py
@@ -0,0 +1,31 @@
+#
+#  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 pywy.core import Executor
+from pywy.core import Plugin
+from pywy.platforms.jvm.execution import JVMExecutor
+from pywy.platforms.jvm.mappings import JVM_OPERATOR_MAPPINGS
+from pywy.platforms.jvm.platform import JVMPlatform
+
+
+class JVMPlugin(Plugin):
+
+    def __init__(self):
+        super(JVMPlugin, self).__init__({JVMPlatform()}, JVM_OPERATOR_MAPPINGS)
+
+    def get_executor(self) -> Executor:
+        return JVMExecutor()
diff --git a/python/src/pywy/platforms/jvm/serializable/__init__.py b/python/src/pywy/platforms/jvm/serializable/__init__.py
new file mode 100644
index 0000000..d9e26de
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/serializable/__init__.py
@@ -0,0 +1,16 @@
+#
+#  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.
+#
diff --git a/python/src/pywy/platforms/jvm/serializable/wayang_jvm_operator.py b/python/src/pywy/platforms/jvm/serializable/wayang_jvm_operator.py
new file mode 100644
index 0000000..d18f155
--- /dev/null
+++ b/python/src/pywy/platforms/jvm/serializable/wayang_jvm_operator.py
@@ -0,0 +1,96 @@
+#
+#  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 typing import Callable, List, TypeVar
+
+from pywy.exception import PywyException
+
+
+class WayangJVMOperator:
+
+    kind: str
+    name: str
+    path: str
+    udf: Callable
+
+    previous: List['WayangJVMOperator']
+    nexts: List['WayangJVMOperator']
+
+    def __init__(self, kind, name):
+        self.name = name
+        self.kind = kind
+        self.previous = []
+        self.nexts = []
+
+    def validate_vector(self, vect: List['WayangJVMOperator'], index: int, op: 'WayangJVMOperator' = None):
+        if op is None:
+            op = self
+
+        if vect is None or len(vect) == 0:
+            vect = [None] * (index + 1)
+
+        if len(vect) < index:
+            vect.extend([None for i in range(index + 1 - len(vect))])
+
+        if vect[index] is not None:
+            raise PywyException(
+                'the position in the index "{}" is already in use for "{}" in the operator "{}"'.format(
+                    index,
+                    vect[index],
+                    op
+                )
+            )
+
+        return vect
+
+    def connect_to(self, nexts_index: int, operator: 'WayangJVMOperator', previous_index: int) -> 'WayangJVMOperator':
+        operator.previous = self.validate_vector(operator.previous, previous_index, operator)
+        self.nexts = self.validate_vector(self.nexts, nexts_index)
+
+        self.nexts[nexts_index] = operator
+        operator.previous[previous_index] = self
+        return self
+
+    def __str__(self):
+        return "WayangJVMOperator {}, previous.[{}], nexts.[{}]".format(
+            self.name,
+            self.previous,
+            self.nexts
+        )
+
+WJO_T = TypeVar('WJO_T', bound=WayangJVMOperator)
+
+
+class WayangJVMMappartitionOperator(WayangJVMOperator):
+
+    def __init__(self, name: str, udf: Callable = None):
+        super().__init__("MapPartitionOperator", name)
+        self.udf = udf
+
+
+class WayangJVMTextFileSource(WayangJVMOperator):
+
+    def __init__(self, name: str, path: str):
+        super().__init__("TextFileSource", name)
+        self.path = path
+
+
+class WayangJVMTextFileSink(WayangJVMOperator):
+
+    def __init__(self, name: str, path: str):
+        super().__init__("TextFileSink", name)
+        self.path = path
+
diff --git a/python/src/pywy/plugins.py b/python/src/pywy/plugins.py
index e3509b6..777d297 100644
--- a/python/src/pywy/plugins.py
+++ b/python/src/pywy/plugins.py
@@ -17,6 +17,7 @@
 
 from pywy.core.platform import Platform
 from pywy.core import Plugin
+from pywy.platforms.jvm.plugin import JVMPlugin
 from pywy.platforms.python.plugin import PythonPlugin
 
 # define the basic plugins that can be used
@@ -25,3 +26,4 @@
 FLINK = Plugin({Platform('flink')})
 # plugin for the python platform
 PYTHON = PythonPlugin()
+JVMs = JVMPlugin()
diff --git a/python/src/pywy/tests/integration/jvm_platform_test.py b/python/src/pywy/tests/integration/jvm_platform_test.py
new file mode 100644
index 0000000..987e733
--- /dev/null
+++ b/python/src/pywy/tests/integration/jvm_platform_test.py
@@ -0,0 +1,144 @@
+#
+#  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
+import os
+import unittest
+import tempfile
+from itertools import chain
+from typing import List, Iterable
+
+from pywy.config import RC_TEST_DIR as ROOT
+from pywy.dataquanta import WayangContext
+from pywy.plugins import JVMs
+
+logger = logging.getLogger(__name__)
+
+
+class TestIntegrationJVMPlatform(unittest.TestCase):
+
+    file_10e0: str
+
+    def setUp(self):
+        self.file_10e0 = "{}/10e0MB.input".format(ROOT)
+
+    @staticmethod
+    def seed_small_grep(validation_file):
+        def pre(a: str) -> bool:
+            return 'six' in a
+
+        fd, path_tmp = tempfile.mkstemp()
+
+        dq = WayangContext() \
+            .register(JVMs) \
+            .textfile(validation_file) \
+            .filter(pre)
+
+        return dq, path_tmp, pre
+
+    def validate_files(self,
+                       validation_file,
+                       outputed_file,
+                       read_and_convert_validation,
+                       read_and_convert_outputed,
+                       delete_outputed=True,
+                       print_variable=False):
+        lines_filter: List[int]
+        with open(validation_file, 'r') as f:
+            lines_filter = list(read_and_convert_validation(f))
+            selectivity = len(lines_filter)
+
+        lines_platform: List[int]
+        with open(outputed_file, 'r') as fp:
+            lines_platform = list(read_and_convert_outputed(fp))
+            elements = len(lines_platform)
+
+        if delete_outputed:
+            os.remove(outputed_file)
+
+        if print_variable:
+            logger.info(f"{lines_platform=}")
+            logger.info(f"{lines_filter=}")
+            logger.info(f"{elements=}")
+            logger.info(f"{selectivity=}")
+
+        self.assertEqual(selectivity, elements)
+        self.assertEqual(lines_filter, lines_platform)
+
+    def test_grep(self):
+
+        dq, path_tmp, pre = self.seed_small_grep(self.file_10e0)
+
+        dq.store_textfile(path_tmp)
+
+        def convert_validation(file):
+            return filter(pre, file.readlines())
+
+        def convert_outputed(file):
+            return file.readlines()
+
+        self.validate_files(
+            self.file_10e0,
+            path_tmp,
+            convert_validation,
+            convert_outputed
+        )
+
+    # def test_dummy_map(self):
+    #
+    #     def convert(a: str) -> int:
+    #         return len(a)
+    #
+    #     dq, path_tmp, pre = self.seed_small_grep(self.file_10e0)
+    #
+    #     dq.map(convert) \
+    #         .store_textfile(path_tmp)
+    #
+    #     def convert_validation(file):
+    #         return map(convert, filter(pre, file.readlines()))
+    #
+    #     def convert_outputed(file):
+    #         return map(lambda x: int(x), file.read().splitlines())
+    #
+    #     self.validate_files(
+    #         self.file_10e0,
+    #         path_tmp,
+    #         convert_validation,
+    #         convert_outputed
+    #     )
+    #
+    # def test_dummy_flatmap(self):
+    #     def fm_func(string: str) -> Iterable[str]:
+    #         return string.strip().split(" ")
+    #
+    #     dq, path_tmp, pre = self.seed_small_grep(self.file_10e0)
+    #
+    #     dq.flatmap(fm_func) \
+    #         .store_textfile(path_tmp, '\n')
+    #
+    #     def convert_validation(file):
+    #         return chain.from_iterable(map(fm_func, filter(pre, file.readlines())))
+    #
+    #     def convert_outputed(file):
+    #         return file.read().splitlines()
+    #
+    #     self.validate_files(
+    #         self.file_10e0,
+    #         path_tmp,
+    #         convert_validation,
+    #         convert_outputed
+    #     )