Python API - Specialized Pipes for SnowFlake, BigQuery etc.

assembly/src/assembly/dist.xml

@@ -26,7 +26,7 @@
     </fileSet>
     <fileSet>
       <directory>${project.parent.basedir}/api</directory>
-      <outputDirectory>api</outputDirectory>
+      <outputDirectory>scala</outputDirectory>
       <includes>
         <include>**/*</include>
       </includes>

api/python/zingg.py → python/api/zingg.py

@@ -71,14 +71,16 @@ def getArgs(self):
     def setArgs(self, argumentsObj):
         self.args = argumentsObj
 
-    def setData(self, pipe):
-        dataPipe = gateway.new_array(jvm.zingg.client.pipe.Pipe, 1)
-        dataPipe[0] = pipe.getPipe()
+    def setData(self, *pipes):
+        dataPipe = gateway.new_array(jvm.zingg.client.pipe.Pipe, len(pipes))
+        for idx, pipe in enumerate(pipes):
+            dataPipe[idx] = pipe.getPipe()
         self.args.setData(dataPipe)
 
-    def setOutput(self, pipe):
-        outputPipe = gateway.new_array(jvm.zingg.client.pipe.Pipe, 1)
-        outputPipe[0] = pipe.getPipe()
+    def setOutput(self, *pipes):
+        outputPipe = gateway.new_array(jvm.zingg.client.pipe.Pipe, len(pipes))
+        for idx, pipe in enumerate(pipes):
+            outputPipe[idx] = pipe.getPipe()
         self.args.setOutput(outputPipe)
 
     def setModelId(self, id):
@@ -163,6 +165,9 @@ def addProperty(self, name, value):
     def setSchema(self, s):
         self.pipe.setSchema(s)
 
+    def toString(self):
+        return self.pipe.toString()
+
 def parseArguments(argv):
     parser = argparse.ArgumentParser(description='Zingg\'s python APIs')
     mandatoryOptions = parser.add_argument_group('mandatory arguments')

python/api/zinggPipes.py

@@ -0,0 +1,56 @@
+import logging
+from zingg import *
+import pandas as pd
+
+LOG = logging.getLogger("zingg.pipes")
+
+Format = jvm.zingg.client.pipe.Format
+FilePipe = jvm.zingg.client.pipe.FilePipe
+
+class CsvPipe(Pipe):
+    def __init__(self, name):
+        Pipe.__init__(self, name, Format.CSV.type())
+        Pipe.addProperty(self, FilePipe.HEADER,"true")
+    def setDelimiter(self, delimiter):
+        Pipe.addProperty(self, "delimiter", delimiter)
+    def setLocation(self, location):
+        Pipe.addProperty(self, FilePipe.LOCATION, location)
+
+class BigQueryPipe(Pipe):
+    def __init__(self,name):
+        Pipe.__init__(self, name, "bigquery")
+        Pipe.addProperty(self, "viewsEnabled", "true")
+    def setCredentialFile(self, file):
+        Pipe.addProperty(self, "credentialsFile", file)
+    def setTable(self, table):
+        Pipe.addProperty(self, "table", table)
+    def setTemporaryGcsBucket(self, bucket):
+        Pipe.addProperty(self, "temporaryGcsBucket", bucket)
+    def setViewsEnabled(self, isEnabled):
+        Pipe.addProperty(self, "viewsEnabled", isEnabled)
+
+class SnowflakePipe(Pipe):
+    def __init__(self,name):
+        Pipe.__init__(self, name, Format.SNOWFLAKE.type())
+    def setURL(self, url):
+        Pipe.addProperty(self, "sfUrl", url)
+    def setUser(self, user):
+        Pipe.addProperty(self, "sfUser", user)
+    def setPassword(self, passwd):
+        Pipe.addProperty(self, "sfPassword", passwd)
+    def setDatabase(self, db):
+        Pipe.addProperty(self, "sfDatabase", db)
+    def setSFSchema(self, schema):
+        Pipe.addProperty(self, "sfSchema", schema)
+    def setWarehouse(self, warehouse):
+        Pipe.addProperty(self, "sfWarehouse", warehouse)
+    def setDbTable(self, dbtable):
+        Pipe.addProperty(self, "dbtable", dbtable)
+
+class InMemoryPipe(Pipe):
+    def __init__(self, name):
+        Pipe.__init__(self, name, Format.INMEMORY.type())
+    def setDataset(self, ds):
+        Pipe.getPipe(self).setDataset(ds)
+    def getDataset(self):
+        return Pipe.getPipe(self).getDataset()

api/python/FebrlExample.py → python/examples/FebrlExample.py

@@ -1,4 +1,4 @@
-from zingg import *
+from python.api.zingg import *
 
 #build the arguments for zingg
 args = Arguments()

python/phases/assessModel.py

@@ -1,4 +1,4 @@
-from api.python.zingg import *
+from python.api.zingg import *
 import pandas as pd
 import seaborn as sn
 import matplotlib.pyplot as plt

python/phases/exportModel.py

@@ -1,5 +1,5 @@
 
-from api.python.zingg import *
+from python.api.zingg import *
 import sys
 import argparse
 import os