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

Question

我test在MySQL中有一个表,其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文件(与测试表相同的数据)和覆盖来尝试相同的程序.这对我有用.

我在这里错过了什么吗？

谢谢!

Answer 1

问题出在您的代码中.因为您覆盖了一个您尝试读取的表,所以在Spark实际访问它之前,您会有效地删除所有数据.

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

你需要的是这样的:

• 创建一个Dataset.

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

• TRUNCATE原始输入和INSERT INTO ... SELECT来自中间表或DROP原始表和RENAME中间表.

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

• 创建一个Dataset.
• 应用所需的转换并将数据写入持久性Spark表(df.write.saveAsTable(...)或等效表)
• TRUNCATE 原始输入.
• 读取数据并保存(spark.table(...).write.jdbc(...))
• Drop Spark表.

我们不能强调使用Spark cache/ persist不是要走的路.即使在保守的StorageLevel(MEMORY_AND_DISK_2/ MEMORY_AND_DISK_SER_2)缓存数据中也可能丢失(节点故障),导致无声的正确性错误.

Answer 2

我认为以上所有步骤都是不必要的。您需要执行以下操作：

• 创建一个数据集，A例如val A = spark.read.parquet("....")

• 读取要更新的表，作为 dataframe B。确保为 dataframe 启用了启用缓存B。val B = spark.read.jdbc("mytable").cache

• 强制count打开B- 这将强制执行并根据所选内容缓存表StorageLevel-B.count

• 现在，您可以进行类似的转换val C = A.union(B)

• 然后写C回数据库，例如C.write.mode(SaveMode.Overwrite).jdbc("mytable")