mysql - Spark SQL和MySQL- SaveMode.Overwrite不插入修改的数据

我在MySQL中有一个test表，其ID和名称如下:

+----+-------+
| id | name  |
+----+-------+
| 1  | Name1 |
+----+-------+
| 2  | Name2 |
+----+-------+
| 3  | Name3 |
+----+-------+

我正在使用Spark DataFrame读取此数据(使用JDBC)并像这样修改数据

Dataset<Row> modified = sparkSession.sql("select id, concat(name,' - new') as name from test");
modified.write().mode("overwrite").jdbc(AppProperties.MYSQL_CONNECTION_URL,
                "test", connectionProperties);

但是我的问题是，如果我提供覆盖模式，它将删除前一个表并创建一个新表，但不会插入任何数据。

我通过读取csv文件(与测试表相同的数据)并覆盖来尝试了相同的程序。那对我有用。

我在这里想念什么吗？

谢谢你!

最佳答案

问题出在您的代码中。因为您覆盖了要从中读取的表，所以实际上可以消除所有数据，然后Spark才能实际访问它。

请记住，Spark是懒惰的。创建Dataset时，Spark会获取所需的元数据，但不会加载数据。因此，没有魔术缓存可以保留原始内容。实际需要时将加载数据。在这里，当您执行write操作并且开始写入时，不再有要提取的数据。

您需要的是这样的:

创建一个Dataset。

应用所需的转换并将数据写入中间MySQL表。

原始表的

TRUNCATE和中间表的原始INSERT INTO ... SELECT或原始表的DROP和中间表的RENAME。

替代方法，但不太有利的方法是:

创建一个Dataset。

应用所需的转换并将数据写入持久性Spark表(df.write.saveAsTable(...)或等效表)

TRUNCATE原始输入。

读回数据并保存(spark.table(...).write.jdbc(...))

删除Spark表。

我们不能过分强调使用Spark cache/persist并不是路要走。即使使用保守的StorageLevel(MEMORY_AND_DISK_2/MEMORY_AND_DISK_SER_2)，缓存的数据也可能会丢失(节点故障)，从而导致无声的正确性错误。