小编典典

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

mysql

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文件(与测试表相同的数据)并覆盖来尝试相同的程序。那对我有用。

我在这里想念什么吗?

谢谢!


阅读 1767

收藏
2020-05-17

共1个答案

小编典典

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

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

您需要的是这样的:

  • 创建一个Dataset
  • 应用所需的转换并将数据写入中间MySQL表。

  • TRUNCATE原始输入以及INSERT INTO ... SELECT来自中间表或DROP原始表与RENAME中间表的输入。

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

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

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

2020-05-17