You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
75 lines
2.3 KiB
75 lines
2.3 KiB
import basic.program
|
|
import utils.config_tool
|
|
import utils.db_abstract
|
|
import pyspark
|
|
import basic.constants
|
|
|
|
DATA_NODE_HEADER = basic.constants.DATA_NODE_HEADER
|
|
DATA_NODE_DATA = basic.constants.DATA_NODE_DATA
|
|
|
|
class DbFcts(utils.db_abstract.DbFcts):
|
|
"""
|
|
This interface defines each necessary connection to any kind of database.
|
|
The specific technique how to connect to the concrete DBMS has to be implemented in the specific tool.
|
|
"""
|
|
def __init__(self):
|
|
pass
|
|
def selectRows(self, table):
|
|
""" method to select rows from a database
|
|
statement written in sql """
|
|
tdata={}
|
|
job = basic.program.Job.getInstance()
|
|
verify = -1+job.getDebugLevel("db_tool")
|
|
cmd = "SELECT "+self.comp.conf["ddl"][table]["fields"].join(",")\
|
|
+" FROM "+table+""+self.getWhere()+""+self.getOrder()";"
|
|
spark = self.getConnector()
|
|
df = spark.sql(cmd)
|
|
data = []
|
|
for r in df:
|
|
data.append(r)
|
|
tdata[DATA_NODE_HEADER] = self.comp.conf["ddl"][table]["fields"]
|
|
tdata[DATA_NODE_DATA] = data
|
|
return tdata
|
|
|
|
def deleteRows(self, table):
|
|
""" method to delete rows from a database
|
|
statement written in sql """
|
|
job = basic.program.Job.getInstance()
|
|
verify = -1+job.getDebugLevel("db_tool")
|
|
cmd = "DELETE FROM "+table+";"
|
|
self.comp.m.logInfo(cmd)
|
|
|
|
def updateRows(self, statement):
|
|
""" method to delete rows from a database
|
|
statement written in sql """
|
|
raise Exception("method is not implemented")
|
|
|
|
def insertRows(self, table, rows):
|
|
""" method to insert rows into a database
|
|
the rows will be interpreted by the ddl of the component
|
|
"""
|
|
job = basic.program.Job.getInstance()
|
|
verify = -1+job.getDebugLevel("db_tool")
|
|
spark = self.getConnector()
|
|
df = spark.createDataFrame(rows)
|
|
|
|
self.comp.m.logInfo("cmd")
|
|
|
|
def getConnector(self):
|
|
""" add-on-method to get the connector
|
|
this method should only called by the class itself """
|
|
job = basic.program.Job.getInstance()
|
|
spark = pyspark.SparkSession.builder()\
|
|
.master("local[1]")\
|
|
.appName("SparkByExamples.com")\
|
|
.getOrCreate()
|
|
return spark
|
|
|
|
@staticmethod
|
|
def execStatement(self, comp, conn, statement):
|
|
""" add-on-method to execute the statement
|
|
this method should only called by the class itself """
|
|
raise Exception("method is not implemented")
|
|
|
|
|
|
|
|
|