[REVIEW-ONLY] Python binding for high level apis by patilvikram · Pull Request #613 · apache/apex-malhar

patilvikram · 2017-04-27T03:42:24Z

I have created this REVIEW-ONLY PR for extending Python support for extending High Level APIs. Currently this is in early stage so I will add documentations which will help you setup environment to launch python app from your machine. Also I will include simple example in this PR.

pramodin · 2017-06-12T15:46:21Z

contrib/src/test/java/com/datatorrent/contrib/hbase/HBaseTransactionalPutOperatorTest.java

@@ -21,12 +21,13 @@
 import java.io.IOException;


Why is this change here? Can you rebase?

pramodin · 2017-06-12T15:50:44Z

python/pom.xml

+        </executions>
+      </plugin>
+       <plugin>
+          <artifactId>maven-dependency-plugin</artifactId>


Does it need to be redefined? Can it not be inherited from the parent.

Fixed this one

pramodin · 2017-06-12T15:51:27Z

python/pom.xml

+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-client</artifactId>
+      <version>${hadoop.version}</version>
+      <!--<scope>provided</scope>-->


Why are these not provided?

pramodin · 2017-06-12T15:52:22Z

python/examples/stock_filter.py

@@ -0,0 +1,14 @@
+


Can this go into standard malhar/examples?

pramodin · 2017-06-12T15:54:19Z

There are some binary files checked in, example py4j-0.10.4.jar, py4j-0.10.4-src.zip, py4j-0.10.4.tar.gz. Are these needed?

pramodin · 2017-06-12T15:56:56Z

stream/src/main/java/org/apache/apex/malhar/python/operator/PythonFilterOperator.java

+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.apex.malhar.python.operator;


Package path doesn't have stream. What is the benefit of having this class here instead of in python package.

pramodin · 2017-06-12T15:57:45Z

stream/src/main/java/org/apache/apex/malhar/python/operator/PythonGenericOperator.java

+    int pythonServerStartAttempts = 5;
+    while (!this.py4jListener.isPythonServerStarted() && !this.pythonWorkerProxy.isFunctionEnabled() && pythonServerStartAttempts > 0) {
+      try {
+        Thread.sleep(5000L);


Timeout should be configurable

pramodin · 2017-06-12T15:58:15Z

stream/src/main/java/org/apache/apex/malhar/python/operator/PythonGenericOperator.java

+        Thread.sleep(5000L);
+        LOG.debug("Waiting for Python Worker Registration");
+        --pythonServerStartAttempts;
+      } catch (InterruptedException var9) {


Variable name.. sounds generated

pramodin · 2017-06-12T15:58:35Z

stream/src/main/java/org/apache/apex/malhar/python/operator/PythonGenericOperator.java

+      this.pythonWorkerProxy.setFunction(this.operationType.getType());
+    }
+    if (!this.py4jListener.isPythonServerStarted()) {
+      LOG.error("Python server could not be started");


Should this throw an exception and fail

pramodin · 2017-06-12T15:59:31Z

stream/src/main/java/org/apache/apex/malhar/stream/api/ApexStream.java

   * @return new stream of type O
   */
-  <O, STREAM extends ApexStream<O>> STREAM addOperator(Operator op, Operator.InputPort<T> inputPort,  Operator.OutputPort<O> outputPort, Option... opts);
+  <O, STREAM extends ApexStream<O>> STREAM addOperator(Operator op, Operator.InputPort<T> inputPort, Operator.OutputPort<O> outputPort, Option... opts);


Avoid formatting change if there is no other change here.

pramodin · 2017-06-12T16:00:16Z

stream/src/main/java/org/apache/apex/malhar/stream/api/ApexStream.java

+   * @param serializedFunction stores Serialized Function data
+   * @return new stream of type T
+   */
+  <STREAM extends ApexStream<T>> STREAM map_func(byte[] serializedFunction, Option... opts);


Function name not in java style

pramodin · 2017-06-12T16:00:46Z

stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexStreamImpl.java

    }
-  }

+  }


Formatting change may be unnecessary

pramodin · 2017-06-12T16:01:51Z

stream/src/main/java/org/apache/apex/malhar/stream/api/util/NetworkUtils.java

+      foundPort = serverSocket.getLocalPort();
+      serverSocket.close();
+      return foundPort;
+    } catch (IOException e) {


finally to close socket

tushargosavi

There are few binaries files present in the pull request, need to discuss with community about which types of binary files are acceptable in repository.

tushargosavi · 2017-06-12T17:22:03Z

library/src/main/java/com/datatorrent/lib/io/fs/InMemoryDataInputOperator.java

+
+public class InMemoryDataInputOperator<T> implements InputOperator
+{
+


extra space

tushargosavi · 2017-06-12T17:22:05Z

docs/python/main.md

+
+from pyapex import createApp
+a=createApp('python_app').fromKafka08('localhost:2181','test_topic') \
+  .setFilter('filter_operator',f) \


I would prefer functions names such as map/filter/flatmap rather than setMap, setFilter, setFlatmap.

tushargosavi · 2017-06-12T17:22:07Z

docs/python/main.md

+
+```
+
+def f(a):


Give meaningful names to functions in documentations. how about filterCondition?

tushargosavi · 2017-06-12T17:22:29Z

library/src/main/java/com/datatorrent/lib/io/fs/InMemoryDataInputOperator.java

+  @Override
+  public void beginWindow(long l)
+  {
+


extra space

tushargosavi · 2017-06-12T17:25:07Z

pom.xml

    <groupId>org.apache.apex</groupId>
    <artifactId>apex</artifactId>
-    <version>3.4.0</version>
+    <version>3.6.0</version>


Apache Apex dependency is already moved to 3.6, you can undo this change.

tushargosavi · 2017-06-12T18:08:57Z

python/src/main/java/org/apache/apex/malhar/python/runtime/PythonWorkerContext.java

+    } else {
+      pythonEnvPath = py4jDependencyFile.getAbsolutePath();
+    }
+    LOG.debug("Final python environment path with Py4j depenency path: " + pythonEnvPath);


use {} in LOG statements

tushargosavi · 2017-06-12T18:09:03Z

python/src/main/java/org/apache/apex/malhar/python/runtime/PythonWorker.java

+
+import java.util.Map;
+
+public interface PythonWorker<T>


please add java documentation for public interfaces / classes and methods

tushargosavi · 2017-06-12T18:11:31Z

stream/pom.xml

      <version>3.2.1</version>
    </dependency>
-
+    <dependency>


remove these dependencies from stream package.

tushargosavi · 2017-06-12T18:11:58Z

stream/src/main/java/org/apache/apex/malhar/stream/api/ApexStream.java

   */
  WindowedStream<T> window(WindowOption windowOption, TriggerOption triggerOption, Duration allowLateness);

+


extra lines added

tushargosavi · 2017-06-12T18:13:27Z

stream/src/test/java/org/apache/apex/malhar/stream/api/impl/ApexStreamImplTest.java

      Assert.assertEquals("first", stream.getSource().getOperatorMeta().getName());
      Assert.assertTrue(1 == stream.getSinks().size());
-      Assert.assertEquals("second", stream.getSinks().get(0).getOperatorWrapper().getName());
+      Assert.assertEquals("second", ((List<LogicalPlan.InputPortMeta>)(stream.getSinks())).get(0).getOperatorMeta().getName());


is this part of your code change?

tushargosavi · 2017-06-12T18:15:25Z

@patilvikram please make sure that jenkins / travis build passes.

patilvikram · 2017-07-13T03:19:37Z

python/apex-python/src/pyapex/functions/window.py

+  def __init__(self, trigger_type, count):
+    super(CountTrigger, self).__init__(trigger_type)
+    self.count = count
+


Remove Extra lines from bottom of this file

Add documentation to code

patilvikram · 2017-07-13T03:27:27Z

python/apex-python/src/setup.py

+setup(name='pyapex',
+      version='0.0.4',
+      py_modules=['pyapex','pyapex.runtime','pyapex.functions'],
+     )


patilvikram · 2017-07-13T03:47:44Z

python/scripts/pyshell

+fi
+if [ -f "$MVN_GENERATED_PATH" ]; then
+  # development launch mode
+  DT_CORE_JAR="$BUILD_DIR/malhar-python-3.8.0-SNAPSHOT.jar"


Make this versions dyanmically stamped

Check how I can include current artifact in the classpath files like MVN_GENERATED_PATH

patilvikram · 2017-07-13T03:48:18Z

python/scripts/pyshell

+#ln -sf $PYAPEX_HOME "$script_dir/pyapex"
+if [ -z "$PYTHONPATH" ] 
+then
+  export PYTHONPATH="$PYAPEX_HOME/deps/pyapex-0.0.4-src.zip"


Add versions dynamically here

patilvikram · 2017-07-13T03:59:26Z

python/src/main/java/org/apache/apex/malhar/python/operator/PythonGenericOperator.java

+{
+  private static final Logger LOG = LoggerFactory.getLogger(PythonGenericOperator.class);
+  protected byte[] serializedFunction = null;
+  private PythonServer server = null;


Make Python Server transient

Fixed this one

patilvikram · 2017-07-13T04:00:28Z

python/src/main/java/org/apache/apex/malhar/python/operator/PythonKeyedWindowedOperator.java

+{
+
+  private static final Logger LOG = LoggerFactory.getLogger(PythonWindowedOperator.class);
+  private PythonServer server = null;


Make PythonServer server transient

patilvikram · 2017-07-13T16:10:15Z

python/apex-python/src/pyapex/commands.py

+      assert False, "Invalid Option"
+
+
+if __name__ == "__main__":


Update this one in pyshell to call apexcli commands directly for kill / shutdown or any other updates

darrengarvey · 2018-12-13T12:05:55Z

Did this branch die? Has it been superceded by another?

I got here from this ticket which was linked from the roadmap.

patilvikram · 2018-12-13T16:28:50Z

Did this branch die? Has it been superceded by another?

I got here from this ticket which was linked from the roadmap.

@darrengarvey Right now its still at the dangling stage, the review is not yet completed. Let me know if you have any questions about this issue or PR.

pramodin reviewed Jun 12, 2017

View reviewed changes

python/examples/stock_filter.py Outdated

@@ -0,0 +1,14 @@

Copy link

Contributor

pramodin Jun 12, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can this go into standard malhar/examples?

pramodin reviewed Jun 12, 2017

View reviewed changes

stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexStreamImpl.java

}

}

}

Copy link

Contributor

pramodin Jun 12, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Formatting change may be unnecessary

pramodin reviewed Jun 12, 2017

View reviewed changes

tushargosavi suggested changes Jun 12, 2017

View reviewed changes

patilvikram force-pushed the python_binding_for_review branch 5 times, most recently from 92a1f1b to cf79a17 Compare June 20, 2017 10:07

patilvikram force-pushed the python_binding_for_review branch from cf79a17 to 151413c Compare June 29, 2017 14:54

tweise mentioned this pull request Jul 6, 2017

[REVIEW-ONLY] Fixed python implementation for Apache Apex #640

Closed

patilvikram force-pushed the python_binding_for_review branch 7 times, most recently from 5336cd4 to a043f96 Compare July 11, 2017 15:28

patilvikram commented Jul 13, 2017

View reviewed changes

Fixed python implementation for Apache Apex

8d06aa2

patilvikram force-pushed the python_binding_for_review branch from 9a7f129 to 8d06aa2 Compare September 11, 2017 13:37

                   }
-                }
+                }


		public class InMemoryDataInputOperator<T> implements InputOperator
		{

		*/
		WindowedStream<T> window(WindowOption windowOption, TriggerOption triggerOption, Duration allowLateness);

Conversation

patilvikram commented Apr 27, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pramodin commented Jun 12, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tushargosavi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tushargosavi commented Jun 12, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

patilvikram Jul 13, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

darrengarvey commented Dec 13, 2018

patilvikram commented Apr 27, 2017 •

edited

Loading

patilvikram Jul 13, 2017 •

edited

Loading