問題描述

我在 MySQL 中有一個(gè) test 表，其 ID 和名稱如下:

+----+-------+|身份證 |姓名 |+----+-------+|1 |姓名1 |+----+-------+|2 |姓名2 |+----+-------+|3 |姓名3 |+----+-------+

我正在使用 Spark DataFrame 讀取此數(shù)據(jù)(使用 JDBC)并像這樣修改數(shù)據(jù)

Datasetmodified = sparkSession.sql("select id, concat(name,' - new') as name from test");modified.write().mode("overwrite").jdbc(AppProperties.MYSQL_CONNECTION_URL,測試"，連接屬性)；

但我的問題是，如果我提供覆蓋模式，它會(huì)刪除以前的表并創(chuàng)建一個(gè)新表但不插入任何數(shù)據(jù).

我通過從 csv 文件(與測試表相同的數(shù)據(jù))讀取并覆蓋來嘗試相同的程序.那對我有用.

我在這里遺漏了什么嗎?

謝謝！

問題出在您的代碼中.因?yàn)槟愀采w了一個(gè)你試圖從中讀取的表，所以在 Spark 可以實(shí)際訪問它之前，你有效地清除了所有數(shù)據(jù).

記住 Spark 是懶惰的.當(dāng)您創(chuàng)建 Dataset 時(shí)，Spark 會(huì)獲取所需的元數(shù)據(jù)，但不會(huì)加載數(shù)據(jù).所以沒有可以保留原始內(nèi)容的魔法緩存.數(shù)據(jù)將在實(shí)際需要時(shí)加載.這是當(dāng)您執(zhí)行 write 操作并且當(dāng)您開始寫入時(shí)沒有更多數(shù)據(jù)要獲取時(shí).

你需要的是這樣的:

• 創(chuàng)建一個(gè)數(shù)據(jù)集.

• 應(yīng)用所需的轉(zhuǎn)換并將數(shù)據(jù)寫入中間 MySQL 表.

• TRUNCATE 原始輸入和 INSERT INTO ... SELECT 來自中間表或 DROP 原始表和 RENAME 中間表.

另一種但不太有利的方法是:

• 創(chuàng)建一個(gè)數(shù)據(jù)集.
• 應(yīng)用所需的轉(zhuǎn)換并將數(shù)據(jù)寫入持久 Spark 表(df.write.saveAsTable(...) 或等效項(xiàng))
• TRUNCATE 原始輸入.
• 讀回?cái)?shù)據(jù)并保存 (spark.table(...).write.jdbc(...))
• 刪除 Spark 表.

我們不能過分強(qiáng)調(diào)使用 Spark cache/persist 不是正確的方法.即使使用保守的 StorageLevel (MEMORY_AND_DISK_2/MEMORY_AND_DISK_SER_2) 緩存數(shù)據(jù)也可能丟失(節(jié)點(diǎn)故障)，導(dǎo)致無提示的正確性錯(cuò)誤.>

I have a test table in MySQL with id and name like below:

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

I am using Spark DataFrame to read this data (using JDBC) and modifying the data like this

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);

But my problem is, if I give overwrite mode, it drops the previous table and creates a new table but not inserting any data.

I tried the same program by reading from a csv file (same data as test table) and overwriting. That worked for me.

Am I missing something here ?

Thank You!

The problem is in your code. Because you overwrite a table from which you're trying to read you effectively obliterate all data before Spark can actually access it.

Remember that Spark is lazy. When you create a Dataset Spark fetches required metadata, but doesn't load the data. So there is no magic cache which will preserve original content. Data will be loaded when it is actually required. Here it is when you execute write action and when you start writing there is no more data to be fetched.

What you need is something like this:

• Create a Dataset.

• Apply required transformations and write data to an intermediate MySQL table.

• TRUNCATE the original input and INSERT INTO ... SELECT from the intermediate table or DROP the original table and RENAME intermediate table.

Alternative, but less favorable approach, would be:

• Create a Dataset.
• Apply required transformations and write data to a persistent Spark table (df.write.saveAsTable(...) or equivalent)
• TRUNCATE the original input.
• Read data back and save (spark.table(...).write.jdbc(...))
• Drop Spark table.

We cannot stress enough that using Spark cache / persist is not the way to go. Even in with the conservative StorageLevel (MEMORY_AND_DISK_2 / MEMORY_AND_DISK_SER_2) cached data can be lost (node failures), leading to silent correctness errors.

這篇關(guān)于Spark SQL 和 MySQL- SaveMode.Overwrite 不插入修改的數(shù)據(jù)的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，也希望大家多多支持html5模板網(wǎng)！