如何处理 Azure DB 的 Databricks 批量插入错误

【问题标题】：How to deal with Databricks Bulk Insert Error to Azure DB如何处理 Azure DB 的 Databricks 批量插入错误
【发布时间】：2019-10-22 18:58:18
【问题描述】：

我正在尝试通过 Azure Databricks 使用 Scala 和 Spark 连接器运行批量插入。我收到来自 SQL Server 的关闭连接错误。一部分数据将传递到目标表，但仅占总数的一小部分。想知道有没有其他人见过这个案例。

编辑：我注意到提到远程 RPC 客户端已解除关联的驱动程序出现错误。这可能是由于击中服务器的批量插入数量已达到阈值上限，我现在正在尝试使用功能较弱的集群来查看降低的并发性是否可以解决问题。

代码：

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

错误转储：

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)

【问题讨论】：

  • 您是否碰巧找到了解决方案？我现在也有同样的问题。谢谢:)

标签： scala azure-sql-database azure-databricks


【解决方案1】：

根据Microsoft SQL Server的官方文档，如下所示，我认为您的问题是由两个配置参数bulkCopyBatchSize和bulkCopyTableLock引起的。

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

在第一个文档中，有一个关于TABLOCK的描述。

TABLOCK：在大容量复制操作期间获取表级锁。此选项显着提高了性能，因为仅在大容量复制操作期间持有锁可减少表上的锁争用。如果表没有索引并且指定了TABLOCK，则可以由多个客户端同时加载表。默认情况下，锁定行为由表选项table lock on bulk load决定。

关键原因在第二个文档中。

批量大小也会影响锁定开销。对 SQL Server 执行大容量复制时，可以使用 bcp_control 指定 TABLOCK 提示来获取表锁而不是行锁。对于整个大容量复制操作，可以以最小的开销保持单个表锁。如果未指定 TABLOCK，则锁会保留在各个行上，并且在大容量复制期间维护所有锁的开销会降低性能。因为锁只在事务的长度内持有，所以指定批处理大小通过定期生成释放当前持有的锁的提交来解决这个问题。

批量复制大量行时，组成批处理的行数会对性能产生显着影响。批量大小的建议取决于正在执行的批量复制的类型。

  • 大容量复制到 SQL Server 时，指定 TABLOCK 大容量复制提示并设置大批量大小。

  • 当未指定 TABLOCK 时，将批处理大小限制为少于 1,000 行。

请注意上面的“当没有指定 TABLOCK 时，将批量大小限制为小于 1,000 行。”，所以我认为你的 Scala 脚本的先前任务已经用完了Azure SQL 数据库的连接池，那么当您将bulkCopyTableLock 值设置为false 并且bulkCopyBatchSize 值大于1000 时，会导致下一个任务无法获得必要的连接。

所以请尝试在你的Scala代码中设置bulkCopyTableLock和true来修复它，甚至适当地减少bulkCopyBatchSize的值。

【讨论】：

    【解决方案2】：

    我建议减少bulkCopyBatchSize的值，这个配置是指每批的行数，而不是字节大小。

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

    这意味着在您当前的设置 "bulkCopyBatchSize" -&gt; "1048576" 下，您每批加载超过 100 万行

    查看 MS 示例：https://github.com/Azure/azure-sqldb-spark/blob/fa1cf19ed797648a20d9b7f474d7c2cd88829ada/samples/scripts/BulkCopySample.scala

    【讨论】：

    • 感谢您的提示。我在表上有一个聚集列存储索引，所以我将批处理大小设置为行数以优化压缩。
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2018-05-25
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2012-07-03
    • 1970-01-01
    相关资源
    最近更新 更多
    热门标签
    Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode