У меня есть таблица на Hive, которая содержит 920 649 653 записей.
Я хотел бы вставить эту таблицу в MS-SQL.
Я использую azure-sqldb-spark lib.
spark2-shell --master = пряжа --jars azure-sqldb-spark-1.0.2-jar-with-dependencies.jar
import com.microsoft.azure.sqldb.spark.bulkcopy.BulkCopyMetadata
import com.microsoft.azure.sqldb.spark.config.Config
import com.microsoft.azure.sqldb.spark.connect._
var bulkCopyMetadata = new BulkCopyMetadata
bulkCopyMetadata.addColumnMetadata(1, "Id_Util_Donnees_Texte", java.sql.Types.INTEGER, 0, 0)
bulkCopyMetadata.addColumnMetadata(2, "Id_Util", java.sql.Types.INTEGER, 0, 0)
bulkCopyMetadata.addColumnMetadata(3, "Id_From", java.sql.Types.INTEGER, 0, 0)
bulkCopyMetadata.addColumnMetadata(4, "IdD", java.sql.Types.INTEGER, 0, 0)
bulkCopyMetadata.addColumnMetadata(5, "Valeur", java.sql.Types.NVARCHAR, 8000, 0)
bulkCopyMetadata.addColumnMetadata(6, "dCollecte", java.sql.Types.TIMESTAMP, 0, 0)
bulkCopyMetadata.addColumnMetadata(7, "dInsertion", java.sql.Types.TIMESTAMP, 0, 0)
bulkCopyMetadata.addColumnMetadata(8, "dMAJ", java.sql.Types.TIMESTAMP, 0, 0)
bulkCopyMetadata.addColumnMetadata(9, "id_marque", java.sql.Types.INTEGER, 0, 0)
val df = spark.table("analyses_tmp.import_sofinco_Util_last_Donnees_Texte").coalesce(10)
val bulkCopyConfig = Config(Map(
"url" -> "db_url",
"user" -> "user",
"password" -> "*******",
"databaseName" -> "Hadoop",
"dbTable" -> "dbo.Util_Last_Donnees_Texte",
"bulkCopyBatchSize" -> "4000",
"bulkCopyTableLock" -> "false",
"bulkCopyTimeout" -> "6000"
))
df.bulkCopyToSqlDB(bulkCopyConfig, bulkCopyMetadata)
Через 2 часа я получил эту ошибку и вся вставка откатилась :
9/04/11 20:17:22 ERROR cluster.YarnClientSchedulerBackend: Yarn application has already exited with state FINISHED!
org.apache.spark.SparkException: Job 1 cancelled because SparkContext was shut down
at org.apache.spark.scheduler.DAGScheduler$$anonfun$cleanUpAfterSchedulerStop$1.apply(DAGScheduler.scala:837)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$cleanUpAfterSchedulerStop$1.apply(DAGScheduler.scala:835)
at scala.collection.mutable.HashSet.foreach(HashSet.scala:78)
at org.apache.spark.scheduler.DAGScheduler.cleanUpAfterSchedulerStop(DAGScheduler.scala:835)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onStop(DAGScheduler.scala:1848)
at org.apache.spark.util.EventLoop.stop(EventLoop.scala:83)
at org.apache.spark.scheduler.DAGScheduler.stop(DAGScheduler.scala:1761)
at org.apache.spark.SparkContext$$anonfun$stop$8.apply$mcV$sp(SparkContext.scala:1931)
at org.apache.spark.util.Utils$.tryLogNonFatalError(Utils.scala:1361)
at org.apache.spark.SparkContext.stop(SparkContext.scala:1930)
at org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend$MonitorThread.run(YarnClientSchedulerBackend.scala:106)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:642)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2034)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2055)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2074)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2099)
at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1.apply(RDD.scala:929)
at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1.apply(RDD.scala:927)
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:363)
at org.apache.spark.rdd.RDD.foreachPartition(RDD.scala:927)
at org.apache.spark.sql.Dataset$$anonfun$foreachPartition$1.apply$mcV$sp(Dataset.scala:2675)
at org.apache.spark.sql.Dataset$$anonfun$foreachPartition$1.apply(Dataset.scala:2675)
at org.apache.spark.sql.Dataset$$anonfun$foreachPartition$1.apply(Dataset.scala:2675)
at org.apache.spark.sql.Dataset$$anonfun$withNewRDDExecutionId$1.apply(Dataset.scala:3239)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
at org.apache.spark.sql.Dataset.withNewRDDExecutionId(Dataset.scala:3235)
at org.apache.spark.sql.Dataset.foreachPartition(Dataset.scala:2674)
at com.microsoft.azure.sqldb.spark.connect.DataFrameFunctions.bulkCopyToSqlDB(DataFrameFunctions.scala:72)
... 51 elided
Видите ли вы причину, по которой произошла эта ошибка?
Есть ли у вас какие-либо рекомендации относительно моих настроек, следует ли мне что-то изменить, чтобы ускорить процесс?