我test在MySQL中有一个表,其ID和名称如下:
+----+-------+ | id | name | +----+-------+ | 1 | Name1 | +----+-------+ | 2 | Name2 | +----+-------+ | 3 | Name3 | +----+-------+
我正在使用Spark DataFrame来读取这些数据(使用JDBC)并修改这样的数据
Dataset 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是懒惰的.创建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)缓存数据中也可能丢失(节点故障),导致无声的正确性错误.
京公网安备 11010802040832号 | 京ICP备19059560号-6 