Merge pull request #559 from aurb9/python-implementation

Expose Parquet operator to Python API

Author: zkaoudi

python/.gitignore

@@ -1,3 +1,3 @@
 # Build files
 dist
-
+__pycache__

python/src/pywy/basic/data/__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.
+#

python/src/pywy/basic/data/record.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.
+#
+
+from typing import Any, List
+
+
+class Record:
+    """
+    A Type that represents a record with a schema.
+    """
+    values: List[Any]
+
+    def __init__(self, values: List[Any]) -> None:
+        self.values = values
+
+    def copy(self) -> 'Record':
+        return Record(self.values.copy())
+
+    def equals(self, o: Any) -> bool:
+        if o is None or type(self) != type(o):
+            return False
+
+        return self.values == o.values
+
+    def hash_code(self) -> int:
+        return hash(self.values)
+
+    def get_field(self, index: int) -> Any:
+        return self.values[index]
+
+    def get_double(self, index: int) -> float:
+        return float(self.values[index])
+
+    def get_int(self, index: int) -> int:
+        return int(self.values[index])
+
+    def get_string(self, index: int) -> str:
+        return str(self.values[index])
+
+    def set_field(self, index: int, field: Any) -> None:
+        self.values[index] = field
+
+    def add_field(self, field: Any) -> None:
+        self.values.append(field)
+
+    def size(self) -> int:
+        return len(self.values)
+
+    def __str__(self):
+        return "Record" + str(self.values)
+
+    def __repr__(self):
+        return self.__str__()

python/src/pywy/core/core.py

@@ -1,4 +1,3 @@
-
 #  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.
@@ -17,18 +16,13 @@
 
 from typing import Set, Iterable, Dict
 import json
-import base64
-import cloudpickle
 import requests
-import subprocess
-import time
-import os
 
 from pywy.core.platform import Platform
 from pywy.core.serializer import JSONSerializer
 from pywy.graph.graph import WayangGraph
-from pywy.graph.types import WGraphOfVec, NodeOperator, NodeVec
-from pywy.operators import SinkOperator, UnaryToUnaryOperator, SourceUnaryOperator
+from pywy.graph.types import WGraphOfVec
+from pywy.operators import SinkOperator, UnaryToUnaryOperator
 
 
 class Plugin:
@@ -122,7 +116,6 @@ def execute(self):
 
         for node in nodes:
             operator = node.current[0]
-
             if isinstance(operator, UnaryToUnaryOperator):
                 pipeline.append(operator)
             else:

python/src/pywy/core/serializer.py

@@ -30,6 +30,7 @@
 from pywy.types import get_java_type, NDimArray, ndim_from_type
 from pywy.operators import SinkOperator, UnaryToUnaryOperator, SourceUnaryOperator
 
+
 class JSONSerializer:
     id_table: Iterable[int]
 
@@ -53,39 +54,42 @@ def serialize(self, operator):
 
         json_operator["data"] = {}
 
-        if operator.json_name != "join":
-            if hasattr(operator, "input_type"):
-                if operator.input_type is not None:
-                    json_operator["data"]["inputType"] = ndim_from_type(operator.input_type).to_json()
-            if hasattr(operator, "output_type"):
-                if operator.output_type is not None:
-                    json_operator["data"]["outputType"] = ndim_from_type(operator.output_type).to_json()
+        if hasattr(operator, "input_type") and operator.input_type is not None:
+            json_operator["data"]["inputType"] = ndim_from_type(operator.input_type).to_json()
+
+        if hasattr(operator, "output_type") and operator.output_type is not None:
+            json_operator["data"]["outputType"] = ndim_from_type(operator.output_type).to_json()
 
         if operator.json_name == "filter":
             json_operator["data"]["udf"] = base64.b64encode(cloudpickle.dumps(operator.use_predicate)).decode('utf-8')
 
-            return json_operator
-        if operator.json_name == "reduceBy":
+        elif operator.json_name == "reduceBy":
             json_operator["data"]["keyUdf"] = base64.b64encode(cloudpickle.dumps(operator.key_function)).decode('utf-8')
             json_operator["data"]["udf"] = base64.b64encode(cloudpickle.dumps(operator.reduce_function)).decode('utf-8')
 
-            return json_operator
         elif operator.json_name == "join":
             json_operator["data"]["thisKeyUdf"] = base64.b64encode(cloudpickle.dumps(operator.this_key_function)).decode('utf-8')
             json_operator["data"]["thatKeyUdf"] = base64.b64encode(cloudpickle.dumps(operator.that_key_function)).decode('utf-8')
 
-            return json_operator
+        elif operator.json_name == "cartesian":
+            del json_operator["data"]
+
         elif operator.json_name == "dlTraining":
             json_operator["data"]["model"] = {"modelType": "DLModel", "op": operator.model.get_out().to_dict()}
             json_operator["data"]["option"] = operator.option.to_dict()
 
-            return json_operator
         else:
             if hasattr(operator, "get_udf"):
                 json_operator["data"]["udf"] = base64.b64encode(cloudpickle.dumps(operator.get_udf)).decode('utf-8')
 
             if hasattr(operator, "path"):
-                json_operator["data"]["filename"] =  operator.path
+                json_operator["data"]["filename"] = operator.path
+
+            if hasattr(operator, "projection"):
+                json_operator["data"]["projection"] = operator.projection
+
+            if hasattr(operator, "column_names"):
+                json_operator["data"]["column_names"] = operator.column_names
 
         return json_operator
 
@@ -98,7 +102,7 @@ def serialize_pipeline(self, pipeline):
         json_operator["cat"] = "unary"
         json_operator["data"] = {}
         json_operator["input"] = list(map(lambda x: self.id_table[x], pipeline[0].inputOperator))
-        json_operator["output"] = list(map(lambda x: self.id_table[x], pipeline[len(pipeline) - 1].outputOperator))
+        json_operator["output"] = list(map(lambda x: self.id_table[x], pipeline[-1].outputOperator))
 
         if len(pipeline) == 1:
             return self.serialize(pipeline[0])

python/src/pywy/dataquanta.py

@@ -15,13 +15,13 @@
 #  limitations under the License.
 #
 
-from typing import Dict, Set, List, cast
+from typing import Dict, Set, List, Optional, cast
 
 from pywy.core.core import Plugin, PywyPlan
 from pywy.operators.base import PO_T
 from pywy.types import (GenericTco, Predicate, Function, BiFunction, FlatmapFunction, IterableOut, T, In, Out)
 from pywy.operators import *
-from pywy.basic.model.ops import Op
+from pywy.basic.data.record import Record
 from pywy.basic.model.option import Option
 from pywy.basic.model.models import Model
 
@@ -50,7 +50,6 @@ def __init__(self, configuration: Configuration = Configuration()):
     """
     add a :class:`Plugin` to the :class:`Context`
     """
-
     def register(self, *plugins: Plugin):
         for p in plugins:
             self.plugins.update(p)
@@ -59,15 +58,19 @@ def register(self, *plugins: Plugin):
     """
     remove a :class:`Plugin` from the :class:`Context`
     """
-
     def unregister(self, *plugins: Plugin):
         for p in plugins:
             self.plugins.remove(p)
         return self
 
-    def textfile(self, file_path: str) -> 'DataQuanta[str]':
+    def textfile(self, file_path: str) -> "DataQuanta[str]":
         return DataQuanta(self, TextFileSource(file_path))
 
+    def parquet(
+        self, file_path: str, projection: Optional[List[str]] = None, column_names: Optional[List[str]] = None
+    ) -> "DataQuanta[Record]":
+        return DataQuanta(self, ParquetSource(file_path, projection, column_names))
+
     def __str__(self):
         return "Plugins: {}".format(str(self.plugins))
 
@@ -88,25 +91,31 @@ def __init__(self, context: WayangContext, operator: PywyOperator):
     def filter(self: "DataQuanta[T]", p: Predicate, input_type: GenericTco = None) -> "DataQuanta[T]":
         return DataQuanta(self.context, self._connect(FilterOperator(p, input_type)))
 
-    def map(self: "DataQuanta[In]", f: Function, input_type: GenericTco = None, output_type: GenericTco = None) -> "DataQuanta[Out]":
+    def map(
+        self: "DataQuanta[In]",
+        f: Function,
+        input_type: GenericTco = None,
+        output_type: GenericTco = None
+    ) -> "DataQuanta[Out]":
         return DataQuanta(self.context, self._connect(MapOperator(f, input_type, output_type)))
 
-    def flatmap(self: "DataQuanta[In]", f: FlatmapFunction, input_type: GenericTco = None, output_type: GenericTco = None) -> "DataQuanta[IterableOut]":
+    def flatmap(
+        self: "DataQuanta[In]",
+        f: FlatmapFunction,
+        input_type: GenericTco = None,
+        output_type: GenericTco = None
+    ) -> "DataQuanta[IterableOut]":
         return DataQuanta(self.context, self._connect(FlatmapOperator(f, input_type, output_type)))
 
-    def reduce_by_key(self: "DataQuanta[In]",
-                      key_f: Function,
-                      f: BiFunction,
-                      input_type: GenericTco = None
-                      ) -> "DataQuanta[IterableOut]":
-
+    def reduce_by_key(
+        self: "DataQuanta[In]",
+        key_f: Function,
+        f: BiFunction,
+        input_type: GenericTco = None
+    ) -> "DataQuanta[IterableOut]":
         return DataQuanta(self.context, self._connect(ReduceByKeyOperator(key_f, f, input_type)))
 
-    def sort(self: "DataQuanta[In]",
-                      key_f: Function,
-                      input_type: GenericTco = None
-                      ) -> "DataQuanta[IterableOut]":
-
+    def sort(self: "DataQuanta[In]", key_f: Function, input_type: GenericTco = None) -> "DataQuanta[IterableOut]":
         return DataQuanta(self.context, self._connect(SortOperator(key_f, input_type)))
 
     def join(
@@ -115,21 +124,34 @@ def join(
         that: "DataQuanta[In]",
         that_key_f: Function,
         input_type: GenericTco = None,
-        output_type: GenericTco = None
-        ) -> "DataQuanta[Out]":
-
+    ) -> "DataQuanta[Out]":
         op = JoinOperator(
             this_key_f,
             that,
             that_key_f,
-            input_type,
-            output_type
+            input_type
         )
 
         self._connect(op),
         return DataQuanta(
             self.context,
-            that._connect(op,1)
+            that._connect(op, 1)
+        )
+
+    def cartesian(
+        self: "DataQuanta[In]",
+        that: "DataQuanta[In]",
+        input_type: GenericTco = None,
+    ) -> "DataQuanta[Out]":
+        op = CartesianOperator(
+            that,
+            input_type
+        )
+
+        self._connect(op),
+        return DataQuanta(
+            self.context,
+            that._connect(op, 1)
         )
 
     def dlTraining(
@@ -140,7 +162,6 @@ def dlTraining(
         input_type: GenericTco,
         output_type: GenericTco
     ) -> "DataQuanta[Out]":
-
         op = DLTrainingOperator(
             model,
             option,
@@ -151,7 +172,7 @@ def dlTraining(
 
         return DataQuanta(
             self.context,
-            that._connect(op,1)
+            that._connect(op, 1)
         )
 
     def predict(
@@ -169,10 +190,10 @@ def predict(
 
         return DataQuanta(
             self.context,
-            that._connect(op,1)
+            that._connect(op, 1)
         )
 
-    def store_textfile(self: "DataQuanta[In]", path: str, input_type: GenericTco = None):
+    def store_textfile(self: "DataQuanta[In]", path: str, input_type: GenericTco = None) -> None:
         last: List[SinkOperator] = [
             cast(
                 SinkOperator,
@@ -184,7 +205,6 @@ def store_textfile(self: "DataQuanta[In]", path: str, input_type: GenericTco = N
                 )
             )
         ]
-        #print(PywyPlan(self.context.plugins, last))
         PywyPlan(self.context.plugins, self.context.configuration.entries, last).execute()
 
     def _connect(self, op: PO_T, port_op: int = 0) -> PywyOperator:

python/src/pywy/execution/util.py

@@ -23,4 +23,3 @@ class SpecialLengths(object):
     END_OF_STREAM = -4
     NULL = -5
     START_ARROW_STREAM = -6
-