我需要在 spark 作业中执行更新查询。我正在尝试下面的代码。但面临问题。import cx_Oracledef query(sql): connection = cx_Oracle.connect("username/password@s<url>/db") cursor = connection.cursor() cursor.execute(sql) result = cursor.fetchall() return resultv = [10]rdd = sc.parallelize(v).coalesce(1)rdd.foreachPartition(lambda x : [query("UPDATE db.tableSET MAPPERS ="+str(i)+" WHERE TABLE_NAME = 'table_name'") for i in x])当我执行上述过程时,出现以下错误。cx_Oracle.InterfaceError: not a query我尝试使用以下代码手动更新。result = query("<update query>")当我这样做时,作业正在连续执行
1 回答
DIEA
TA贡献1820条经验 获得超2个赞
我解决了我的问题。根据卢克的输入。我使用fetchall()which 用于查询。我需要使用commit(). 所以更改了代码并检查其工作正常。
import cx_Oracle
def query(sql):
connection = cx_Oracle.connect("username/password@s<url>/db")
cursor = connection.cursor()
cursor.execute(sql)
result = connection.commit()
v = [10]
rdd = sc.parallelize(v).coalesce(1)
rdd.foreachPartition(lambda x : [query("UPDATE db.tableSET MAPPERS ="+str(i)+"WHERE TABLE_NAME = 'table_name'") for i in x])
添加回答
举报
0/150
提交
取消