使用 INSERT INTO table ON DUPLICATE KEY 时出错，使用 for 循环数组

Question

我正在使用 pyspark 框架更新 mysql 数据库，并在 AWS Glue 服务上运行。

我有一个数据框如下：

df2= sqlContext.createDataFrame([("xxx1","81A01","TERR NAME 55","NY"),("xxx2","81A01","TERR NAME 55","NY"),("x103","81A01","TERR NAME 01","NJ")], ["zip_code","territory_code","territory_name","state"])

# Print out information about this data
df2.show()
+--------+--------------+--------------+-----+
|zip_code|territory_code|territory_name|state|
+--------+--------------+--------------+-----+
|    xxx1|         81A01|  TERR NAME 55|   NY|
|    xxx2|         81A01|  TERR NAME 55|   NY|
|    x103|         81A01|  TERR NAME 01|   NJ|
+---------------------------------------------

我有一个主键 ZIP_CODE，我需要确保没有重复键或主键异常，因此我正在使用 INSERT INTO .... ON DUPLICATE KEYS。

并且因为我有不止一行到 insert/update，我使用 for array in python 来遍历记录，并执行 INSERT 到数据库中。代码如下：

sarry = df2.collect()
for r in sarry:
     db = MySQLdb.connect("xxxx.rds.amazonaws.com", "username", "password", 
      "databasename")
     cursor = db.cursor()
     insertQry=INSERT INTO ZIP_TERR(zip_code, territory_code, territory_name, 
     state) VALUES(r.zip_code, r.territory_code, r.territory_name, r.state) ON 
     DUPLICATE KEY UPDATE territory_name = VALUES(territory_name), state = 
     VALUES(state);"
     n=cursor.execute(insertQry)
     db.commit()
     db.close()

当运行上述插入查询函数时，我得到以下错误信息，无法得到任何关于错误的线索。请帮忙。

Traceback (most recent call last):
  File "/tmp/zeppelin_pyspark-2291407229037300959.py", line 367, in <module>
    raise Exception(traceback.format_exc())
Exception: Traceback (most recent call last):
  File "/tmp/zeppelin_pyspark-2291407229037300959.py", line 360, in <module>
    exec(code, _zcUserQueryNameSpace)
  File "<stdin>", line 8, in <module>
  File "/usr/local/lib/python2.7/site-packages/pymysql/cursors.py", line 170, in execute
    result = self._query(query)
  File "/usr/local/lib/python2.7/site-packages/pymysql/cursors.py", line 328, in _query
    conn.query(q)
  File "/usr/local/lib/python2.7/site-packages/pymysql/connections.py", line 893, in query
    self._affected_rows = self._read_query_result(unbuffered=unbuffered)
  File "/usr/local/lib/python2.7/site-packages/pymysql/connections.py", line 1103, in _read_query_result
    result.read()
  File "/usr/local/lib/python2.7/site-packages/pymysql/connections.py", line 1396, in read
    first_packet = self.connection._read_packet()
  File "/usr/local/lib/python2.7/site-packages/pymysql/connections.py", line 1059, in _read_packet
    packet.check_error()
  File "/usr/local/lib/python2.7/site-packages/pymysql/connections.py", line 384, in check_error
    err.raise_mysql_exception(self._data)
  File "/usr/local/lib/python2.7/site-packages/pymysql/err.py", line 109, in raise_mysql_exception
    raise errorclass(errno, errval)
InternalError: (1054, u"Unknown column 'r.zip_code' in 'field list'")

如果我只是尝试打印一行的值，我得到的打印值如下：

print('zip_code_new: ', r.zip_code, r.territory_code, r.territory_name, r.state)

zip_code_new:  xxx1 81A01 TERR NAME 55 NY

谢谢。我在 AWS Glue/Pyspark 上工作，所以我需要使用本机 python 库。

Answer 1

以下插入查询有效，带有 for 循环。

insertQry="INSERT INTO ZIP_TERR(zip_code, territory_code, territory_name, state) VALUES(%s, %s, %s, %s) ON DUPLICATE KEY UPDATE territory_name = %s, state = %s;

n=cursor.execute(insertQry, (r.zip_code, r.territory_code, r.territory_name, r.state, r.territory_name, r.state))
print (" CURSOR status :", n)

结果输出：

CURSOR status : 2

谢谢。希望这对其他人有参考意义。

使用 INSERT INTO table ON DUPLICATE KEY 时出错，使用 for 循环数组

Error while using INSERT INTO table ON DUPLICATE KEY, using a for loop array

mysql

python-2.7

pyspark

pyspark-sql

aws-glue