不能大DataFrame寫入該服務器通過使用jdbc驅動程序在Azure磚嗎

[ninjadev999]

我閱讀一個巨大的csv文件包括39795158條記錄和寫作到該服務器,在Azure磚。磚(筆記本)運行在集群節點56 GB內存,16個核心,和12個工人。

這是我在Python和PySpark代碼:

從pyspark。sql從pyspark.sql進口*。從pyspark.sql進口*類型。函數從進口睡眠時間進口* url = " jdbc::狀態"置疑" / /{0}:{1};數據庫= {2}”。格式(服務器、端口、數據庫)spark.conf.set (“spark.databricks.io.cache。”,真正的)spark.conf.set (“spark.sql.execution.arrow啟用。啟用”、“真實”)#讀csv文件。df_lake =火花。讀\ .option(“頭”,“假”). schema (s) \ \ .option(“分隔符”,“,”)\ . csv (“wasbs: / /…”) batch_size = 60000行= df_lake.count () org_pts = df_lake.rdd.getNumPartitions () # 566 new_pts = 1990 # Re-partition DataFrame df_repartitioned = df_lake.repartition (new_pts) #寫DataFrame到該服務器,通過使用JDBC驅動程序df_repartitioned。寫\ .format (jdbc) \ .mode(“覆蓋”)\ .option(“司機”,“com.microsoft.sqlserver.jdbc.SQLServerDriver”) \ .option (“url”, url) \ .option(“數據表”,表)\ .option(“用戶”,用戶名)\ .option(“密碼”,密碼)\ .option (“batchsize”, batch_size) \ .save睡眠()(10)

然後我得到了日誌和錯誤如下:

' ' '

行:39795158

org_pts: 566

new_pts: 1990

複製錯誤:調用o9647.save時發生一個錯誤。

:org.apache.spark。SparkException:工作階段失敗而終止:62年任務階段462.0失敗了4次,最近的失敗:在舞台上失去了任務62.3 462.0 (TID 46609)(10.139.64.12執行人27):com.microsoft.sqlserver.jdbc。SQLServerException:連接關閉。

com.microsoft.sqlserver.jdbc.SQLServerException.makeFromDriverError (SQLServerException.java: 234)

com.microsoft.sqlserver.jdbc.SQLServerConnection.checkClosed (SQLServerConnection.java: 1217)

com.microsoft.sqlserver.jdbc.SQLServerConnection.rollback (SQLServerConnection.java: 3508)

org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils .savePartition美元(JdbcUtils.scala: 728)

在org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils。美元anonfun saveTable 1美元(JdbcUtils.scala: 857)

org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils。美元anonfun saveTable美元$ 1 $改編(JdbcUtils.scala: 855)

在org.apache.spark.rdd.RDD。anonfun foreachPartition美元2美元(RDD.scala: 1025)

org.apache.spark.rdd.RDD。anonfun foreachPartition美元$ 2 $改編(RDD.scala: 1025)

在org.apache.spark.SparkContext。anonfun runJob美元2美元(SparkContext.scala: 2517)

在org.apache.spark.scheduler.ResultTask。anonfun runTask美元3美元(ResultTask.scala: 75)

在美元com.databricks.spark.util.ExecutorFrameProfiler知根知底(ExecutorFrameProfiler.scala: 110)

在org.apache.spark.scheduler.ResultTask。anonfun runTask美元1美元(ResultTask.scala: 75)

在美元com.databricks.spark.util.ExecutorFrameProfiler知根知底(ExecutorFrameProfiler.scala: 110)

org.apache.spark.scheduler.ResultTask.runTask (ResultTask.scala: 55)

org.apache.spark.scheduler.Task.doRunTask (Task.scala: 150)

在org.apache.spark.scheduler.Task。anonfun運行$ 1美元(Task.scala: 119)

在美元com.databricks.spark.util.ExecutorFrameProfiler知根知底(ExecutorFrameProfiler.scala: 110)

org.apache.spark.scheduler.Task.run (Task.scala: 91)

在org.apache.spark.executor.Executor TaskRunner。美元anonfun運行13美元(Executor.scala: 813)

org.apache.spark.util.Utils .tryWithSafeFinally美元(Utils.scala: 1620)

在org.apache.spark.executor.Executor TaskRunner。美元anonfun運行4美元(Executor.scala: 816)

在scala.runtime.java8.JFunction0專門sp.apply美元(美元JFunction0 mcV $ sp.java: 23)

在美元com.databricks.spark.util.ExecutorFrameProfiler知根知底(ExecutorFrameProfiler.scala: 110)

org.apache.spark.executor.Executor TaskRunner.run美元(Executor.scala: 672)

java.util.concurrent.ThreadPoolExecutor.runWorker (ThreadPoolExecutor.java: 1149)

java.util.concurrent.ThreadPoolExecutor Worker.run美元(ThreadPoolExecutor.java: 624)

java.lang.Thread.run (Thread.java: 748)

…

' ' '

3 - 6數以百萬計的記錄,這是沒有問題。但10數百萬以上記錄,這是失敗了。我不知道為什麼它是發生在10數百萬以上記錄。

有什麼解決方案巨大DataFrame過程Azure磚嗎?

我也在StackOverflow上傳了這個錯誤。你可以檢查錯誤https://stackoverflow.com/questions/71076648/cant-write-big-dataframe-into-mssql-server-by-using-jdb..。

[werners1]

´我不明白為什麼你使用磚。沒有數據轉換,隻有從csv移動數據到數據庫中。

為什麼´t和數據工廠你這樣做嗎?

事情是這樣的:磚將csv文件沒有處理這個問題,但是,當你將它寫入數據庫,將成為瓶頸。你也有一個集群運行主要是等待。

[ninjadev999]

謝謝你的回應!

事實上,我不知道對數據工廠. .

加快處理大DataFrame,我用再分配模式,根據這個博客:https://devblogs.microsoft.com/azure-sql/partitioning-on-spark-fast-loading-clustered-columnstore-in..。

你能讓我知道我如何使用它來覆蓋所有數據在csv文件到該軟件嗎?是容易的嗎?

[werners1]

數據工廠是非常容易的。基本上你定義一個源數據駐留在何處使用UI。

土地數據定義一個水槽,還使用一個界麵,然後執行工作。

對於普通的數據移動理想。但是如果你必須添加轉換等,我更喜歡磚(你也可以在數據工廠,但我不喜歡無/低代碼工具)。

[ninjadev999]

我應該做出一些改變在每個記錄。之後,我將它們寫入到該軟件。所以看起來像磚是更好的使用它…

我隻是需要加快大dataframe寫入該數據庫並修複錯誤。