apache-spark - Spark jdbc覆盖模式未按预期工作

Question

我想使用 spark 执行更新和插入操作请找到现有表的图像参考

在这里，我正在更新 id :101 location 和 inserttime 并插入另外 2 条记录：

并使用模式覆盖写入目标

df.write.format("jdbc")
  .option("url",  "jdbc:mysql://localhost/test")
  .option("driver","com.mysql.jdbc.Driver")
  .option("dbtable","temptgtUpdate")
  .option("user", "root")
  .option("password", "root")
  .option("truncate","true")
  .mode("overwrite")
  .save()

执行上述命令后，我的数据已损坏，已插入 db 表

数据框中的数据

您能否让我知道您的观察和解决方案

Answer 1

如果您需要在 pyspark 代码中执行 UPSERT/DELETE 操作，我建议您使用 pymysql 库，并执行您的 upsert/delete 操作。请查看此帖子以获取更多信息和代码示例以供参考：在使用重复键上的插入表时出错，使用 for 循环数组

请根据您的需要修改代码示例。

Answer 2

Spark JDBC writer 支持以下模式：

• append：将此 :class:DataFrame 的内容附加到现有数据。
• 覆盖：覆盖现有数据。
• 忽略：如果数据已经存在，则静默忽略此操作。

• 错误（默认情况）：如果数据已经存在则抛出异常

• https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html

由于您使用的是“覆盖”模式，因此它会根据列长度重新创建表，如果您希望自己的表定义首先创建表并使用“附加”模式

Answer 3

我想使用 spark 执行更新和插入操作

Spark SQL中没有与 SQLUPDATE语句等效的功能。也没有与 Spark SQL 等效的 SQLDELETE WHERE语句。相反，您必须在 Spark 之外删除需要更新的行，然后使用模式将包含新记录和更新记录的 Spark 数据帧写入表append（以保留表中剩余的现有行）。

Answer 4

我不推荐 TRUNCATE，因为它实际上会删除表并创建新表。执行此操作时，表可能会丢失之前设置的列级属性......所以在使用 TRUNCATE 时要小心，并确保删除表/重新创建表是否可以。

Answer 5

执行以下步骤时，Upsert 逻辑工作正常

  df = (spark.read.format("csv").
        load("file:///C:/Users/test/Desktop/temp1/temp1.csv", header=True,
             delimiter=','))

并这样做

  (df.write.format("jdbc").
    option("url", "jdbc:mysql://localhost/test").
    option("driver", "com.mysql.jdbc.Driver").
    option("dbtable", "temptgtUpdate").
    option("user", "root").
    option("password", "root").
    option("truncate", "true").
    mode("overwrite").save())

尽管如此，当我直接使用数据框编写时，我仍然无法理解为什么它失败的逻辑