How to deal with Databricks Bulk Insert Error to Azure DB

user_5935979 0
Viewed 2941

I'm trying to run a bulk insert using Scala & the Spark Connector via Azure Databricks. I'm getting closed connection errors from SQL Server. A portion of the data will pass through to the destination table but it's only a small percentage of the total. Wondering if anybody else has seen this case before.

EDIT: I noticed errors on the drivers that mention Remote RPC client disassociated. It may be due to the # of bulk inserts hitting the server are maxing out thresholds, I'm trying now with a less powerful cluster to see if the reduced concurrency fixes it any.

Code:

%scala
import com.microsoft.azure.sqldb.spark.bulkcopy.BulkCopyMetadata
import com.microsoft.azure.sqldb.spark.config.Config
import com.microsoft.azure.sqldb.spark.connect._

val bulkCopyConfig = Config(Map(
  "url"               -> "myserver.database.windows.net",
  "databaseName"      -> "mydb",
  "user"              -> "myuser",
  "password"          -> "mypw",
  "dbTable"           -> "my_sql_tbl",
  "bulkCopyBatchSize" -> "1048576",
  "bulkCopyTableLock" -> "false",
  "loginTimeout"      -> "3600",
  "bulkCopyTimeout"   -> "100000000"
))

spark.table("my_databricks_tbl").bulkCopyToSqlDB(bulkCopyConfig)

Error Dump:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 171 in stage 75.0 failed 4 times, most recent failure: Lost task 171.3 in stage 75.0 (TID 5315, 10.139.64.16, executor 172): com.microsoft.sqlserver.jdbc.SQLServerException: The connection is closed.
    at com.microsoft.sqlserver.jdbc.SQLServerException.makeFromDriverError(SQLServerException.java:227)
    at com.microsoft.sqlserver.jdbc.SQLServerConnection.checkClosed(SQLServerConnection.java:796)
    at com.microsoft.sqlserver.jdbc.SQLServerConnection.rollback(SQLServerConnection.java:2698)
    at com.microsoft.azure.sqldb.spark.connect.DataFrameFunctions.com$microsoft$azure$sqldb$spark$connect$DataFrameFunctions$$bulkCopy(DataFrameFunctions.scala:142)
    at com.microsoft.azure.sqldb.spark.connect.DataFrameFunctions$$anonfun$bulkCopyToSqlDB$1.apply(DataFrameFunctions.scala:72)
    at com.microsoft.azure.sqldb.spark.connect.DataFrameFunctions$$anonfun$bulkCopyToSqlDB$1.apply(DataFrameFunctions.scala:72)
    at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$28.apply(RDD.scala:951)
    at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$28.apply(RDD.scala:951)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2320)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2320)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
    at org.apache.spark.scheduler.Task.doRunTask(Task.scala:140)
    at org.apache.spark.scheduler.Task.run(Task.scala:113)
    at org.apache.spark.executor.Executor$TaskRunner$$anonfun$13.apply(Executor.scala:528)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1526)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:534)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

Driver stacktrace:
    at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:2355)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2343)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2342)
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
    at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2342)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1096)
    at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1096)
    at scala.Option.foreach(Option.scala:257)
    at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1096)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2574)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2522)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2510)
    at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
    at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:893)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2279)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2301)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2320)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2345)
    at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1.apply(RDD.scala:951)
    at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1.apply(RDD.scala:949)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:379)
    at org.apache.spark.rdd.RDD.foreachPartition(RDD.scala:949)
    at org.apache.spark.sql.Dataset$$anonfun$foreachPartition$1.apply$mcV$sp(Dataset.scala:2801)
    at org.apache.spark.sql.Dataset$$anonfun$foreachPartition$1.apply(Dataset.scala:2801)
    at org.apache.spark.sql.Dataset$$anonfun$foreachPartition$1.apply(Dataset.scala:2801)
    at org.apache.spark.sql.Dataset$$anonfun$withNewRDDExecutionId$1.apply(Dataset.scala:3462)
    at org.apache.spark.sql.execution.SQLExecution$$anonfun$withCustomExecutionEnv$1.apply(SQLExecution.scala:111)
    at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:240)
    at org.apache.spark.sql.execution.SQLExecution$.withCustomExecutionEnv(SQLExecution.scala:97)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:170)
    at org.apache.spark.sql.Dataset.withNewRDDExecutionId(Dataset.scala:3458)
    at org.apache.spark.sql.Dataset.foreachPartition(Dataset.scala:2800)
    at com.microsoft.azure.sqldb.spark.connect.DataFrameFunctions.bulkCopyToSqlDB(DataFrameFunctions.scala:72)
    at line12e7f598547340fe98bb640a4cccc5a836.$read$$iw$$iw$$iw$$iw$$iw$$iw.<init>(command-3923847285087761:67)
    at line12e7f598547340fe98bb640a4cccc5a836.$read$$iw$$iw$$iw$$iw$$iw.<init>(command-3923847285087761:164)
    at line12e7f598547340fe98bb640a4cccc5a836.$read$$iw$$iw$$iw$$iw.<init>(command-3923847285087761:166)
    at line12e7f598547340fe98bb640a4cccc5a836.$read$$iw$$iw$$iw.<init>(command-3923847285087761:168)
    at line12e7f598547340fe98bb640a4cccc5a836.$read$$iw$$iw.<init>(command-3923847285087761:170)
    at line12e7f598547340fe98bb640a4cccc5a836.$read$$iw.<init>(command-3923847285087761:172)
    at line12e7f598547340fe98bb640a4cccc5a836.$read.<init>(command-3923847285087761:174)
    at line12e7f598547340fe98bb640a4cccc5a836.$read$.<init>(command-3923847285087761:178)
    at line12e7f598547340fe98bb640a4cccc5a836.$read$.<clinit>(command-3923847285087761)
    at line12e7f598547340fe98bb640a4cccc5a836.$eval$.$print$lzycompute(<notebook>:7)
    at line12e7f598547340fe98bb640a4cccc5a836.$eval$.$print(<notebook>:6)
    at line12e7f598547340fe98bb640a4cccc5a836.$eval.$print(<notebook>)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at scala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:793)
    at scala.tools.nsc.interpreter.IMain$Request.loadAndRun(IMain.scala:1054)
    at scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:645)
    at scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:644)
    at scala.reflect.internal.util.ScalaClassLoader$class.asContext(ScalaClassLoader.scala:31)
    at scala.reflect.internal.util.AbstractFileClassLoader.asContext(AbstractFileClassLoader.scala:19)
    at scala.tools.nsc.interpreter.IMain$WrappedRequest.loadAndRunReq(IMain.scala:644)
    at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:576)
    at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:572)
    at com.databricks.backend.daemon.driver.DriverILoop.execute(DriverILoop.scala:215)
    at com.databricks.backend.daemon.driver.ScalaDriverLocal$$anonfun$repl$1.apply$mcV$sp(ScalaDriverLocal.scala:197)
    at com.databricks.backend.daemon.driver.ScalaDriverLocal$$anonfun$repl$1.apply(ScalaDriverLocal.scala:197)
    at com.databricks.backend.daemon.driver.ScalaDriverLocal$$anonfun$repl$1.apply(ScalaDriverLocal.scala:197)
    at com.databricks.backend.daemon.driver.DriverLocal$TrapExitInternal$.trapExit(DriverLocal.scala:694)
    at com.databricks.backend.daemon.driver.DriverLocal$TrapExit$.apply(DriverLocal.scala:647)
    at com.databricks.backend.daemon.driver.ScalaDriverLocal.repl(ScalaDriverLocal.scala:197)
    at com.databricks.backend.daemon.driver.DriverLocal$$anonfun$execute$9.apply(DriverLocal.scala:381)
    at com.databricks.backend.daemon.driver.DriverLocal$$anonfun$execute$9.apply(DriverLocal.scala:358)
    at com.databricks.logging.UsageLogging$$anonfun$withAttributionContext$1.apply(UsageLogging.scala:241)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at com.databricks.logging.UsageLogging$class.withAttributionContext(UsageLogging.scala:236)
    at com.databricks.backend.daemon.driver.DriverLocal.withAttributionContext(DriverLocal.scala:49)
    at com.databricks.logging.UsageLogging$class.withAttributionTags(UsageLogging.scala:278)
    at com.databricks.backend.daemon.driver.DriverLocal.withAttributionTags(DriverLocal.scala:49)
    at com.databricks.backend.daemon.driver.DriverLocal.execute(DriverLocal.scala:358)
    at com.databricks.backend.daemon.driver.DriverWrapper$$anonfun$tryExecutingCommand$2.apply(DriverWrapper.scala:644)
    at com.databricks.backend.daemon.driver.DriverWrapper$$anonfun$tryExecutingCommand$2.apply(DriverWrapper.scala:644)
    at scala.util.Try$.apply(Try.scala:192)
    at com.databricks.backend.daemon.driver.DriverWrapper.tryExecutingCommand(DriverWrapper.scala:639)
    at com.databricks.backend.daemon.driver.DriverWrapper.getCommandOutputAndError(DriverWrapper.scala:485)
    at com.databricks.backend.daemon.driver.DriverWrapper.executeCommand(DriverWrapper.scala:597)
    at com.databricks.backend.daemon.driver.DriverWrapper.runInnerLoop(DriverWrapper.scala:390)
    at com.databricks.backend.daemon.driver.DriverWrapper.runInner(DriverWrapper.scala:337)
    at com.databricks.backend.daemon.driver.DriverWrapper.run(DriverWrapper.scala:219)
    at java.lang.Thread.run(Thread.java:748)
Caused by: com.microsoft.sqlserver.jdbc.SQLServerException: The connection is closed.
    at com.microsoft.sqlserver.jdbc.SQLServerException.makeFromDriverError(SQLServerException.java:227)
    at com.microsoft.sqlserver.jdbc.SQLServerConnection.checkClosed(SQLServerConnection.java:796)
    at com.microsoft.sqlserver.jdbc.SQLServerConnection.rollback(SQLServerConnection.java:2698)
    at com.microsoft.azure.sqldb.spark.connect.DataFrameFunctions.com$microsoft$azure$sqldb$spark$connect$DataFrameFunctions$$bulkCopy(DataFrameFunctions.scala:142)
    at com.microsoft.azure.sqldb.spark.connect.DataFrameFunctions$$anonfun$bulkCopyToSqlDB$1.apply(DataFrameFunctions.scala:72)
    at com.microsoft.azure.sqldb.spark.connect.DataFrameFunctions$$anonfun$bulkCopyToSqlDB$1.apply(DataFrameFunctions.scala:72)
    at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$28.apply(RDD.scala:951)
    at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$28.apply(RDD.scala:951)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2320)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2320)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
    at org.apache.spark.scheduler.Task.doRunTask(Task.scala:140)
    at org.apache.spark.scheduler.Task.run(Task.scala:113)
    at org.apache.spark.executor.Executor$TaskRunner$$anonfun$13.apply(Executor.scala:528)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1526)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:534)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
azure-databricks azure-sql-database scala
2 Answers

I would suggest reducing the value of bulkCopyBatchSize, this configuration refers to the the number of rows in each batch, and not to a bytes size.

https://static.javadoc.io/com.microsoft.sqlserver/mssql-jdbc/6.1.6.jre8-preview/com/microsoft/sqlserver/jdbc/SQLServerBulkCopyOptions.html#setBatchSize-int-

This means that with your current setting "bulkCopyBatchSize" -> "1048576" you are loading more than 1 million rows per batch

Have a look at the MS sample : https://github.com/Azure/azure-sqldb-spark/blob/fa1cf19ed797648a20d9b7f474d7c2cd88829ada/samples/scripts/BulkCopySample.scala

user_55546660

According to the offical documents for Microsoft SQL Server, as below, I think your issue was caused by two configure parameters bulkCopyBatchSize and bulkCopyTableLock.

  1. Performing Bulk Copy Operations
  2. Managing Bulk Copy Batch Sizes

In the first document, there is a description about TABLOCK.

TABLOCK: A table-level lock is acquired for the duration of the bulk copy operation. This option significantly improves performance because holding a lock only for the duration of the bulk copy operation reduces lock contention on the table. A table can be loaded by multiple clients concurrently if the table has no indexes and TABLOCK is specified. By default, the locking behavior is determined by the table option table lock on bulk load.

And the key reason is in the second doc.

The batch size can also affect locking overhead. When performing a bulk copy against SQL Server, the TABLOCK hint can be specified using bcp_control to acquire a table lock instead of row locks. The single table lock can be held with minimal overhead for an entire bulk copy operation. If TABLOCK is not specified then locks are held on individual rows and the overhead of maintaining all the locks for the duration of the bulk copy can slow performance. Because locks are only held for the length of a transaction, specifying a batch size addresses this problem by periodically generating a commit that frees the locks currently held.

The number of rows making up a batch can have significant performance effects when bulk copying a large number of rows. The recommendations for batch size depend on the type of bulk copy being performed.

  • When bulk copying to SQL Server, specify the TABLOCK bulk copy hint and set a large batch size.

  • When TABLOCK is not specified, limit batch sizes to less than 1,000 rows.

Please notice the words above "When TABLOCK is not specified, limit batch sizes to less than 1,000 rows.", so I think the previous tasks of your Scala script had used up the connections of the connection pool of Azure SQL Database, then to cause the next tasks could not get the necessary connections when you set the bulkCopyTableLock value with false and the bulkCopyBatchSize value more than 1000.

So please try to set bulkCopyTableLock with true to fix it in your Scala code, even reduce the value of bulkCopyBatchSize properly.

user_49896760
Related