与SQL Server的JDBC连接经常超时

Question

我正在通过sparklyr V0.6.2运行Spark V2.2.1并通过JDBC从SQL Server提取数据。我似乎遇到了一些网络问题，因为我的执行程序多次（不是每次）对SQL Server执行写操作失败，并出现错误：

Prelogin error: host <my server> port 1433 Error reading prelogin response: Connection timed out (Read failed) ClientConnectionId:...

我正在使用以下配置运行sparklyr会话：

spark_conf = spark_config()
spark_conf$spark.executor.cores <- 8
spark_conf$`sparklyr.shell.driver-memory` <- "8G"
spark_conf$`sparklyr.shell.executor-memory` <- "12G"
spark_conf$spark.serializer <- "org.apache.spark.serializer.KryoSerializer"
spark_conf$spark.network.timeout <- 400

18/06/11 17:53:44 INFO BlockManager: Found block rdd_9_16 locally
18/06/11 17:53:45 WARN SQLServerConnection: ConnectionID:3 ClientConnectionId: d3568a9f-049f-4772-83d4-ed65b907fc8b Prelogin error: host nciensql14.nciwin.local port 1433 Error reading prelogin response: Connection timed out (Read failed) ClientConnectionId:d3568a9f-049f-4772-83d4-ed65b907fc8b
18/06/11 17:53:45 WARN SQLServerConnection: ConnectionID:2 ClientConnectionId: ecb084e6-99a8-49d1-9215-491324e8d133 Prelogin error: host nciensql14.nciwin.local port 1433 Error reading prelogin response: Connection timed out (Read failed) ClientConnectionId:ecb084e6-99a8-49d1-9215-491324e8d133
18/06/11 17:53:45 ERROR Executor: Exception in task 10.0 in stage 26.0 (TID 77)

function (df, tbl, db, server = NULL, user, pass, mode = "error", 
    options = list(), ...) 
{
    sparklyr::spark_write_jdbc(
  df, 
  tbl, 
  options = c(
    list(url = paste0("jdbc:sqlserver://", server, ".nciwin.local;", 
         "databaseName=", db, ";", 
         "user=", user, ";", 
         "password=", pass, ";"), 
       driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"), 
    options), 
  mode = mode, ...)
}

function (sc, tbl, db, server = NULL, user, pass, repartition = 0, options = list(), ...) 
{
    sparklyr::spark_read_jdbc(
      sc, 
      tbl, 
      options = c(
        list(url = paste0("jdbc:sqlserver://", server, ".nciwin.local;"), 
             user = user, 
             password = pass, 
             databaseName = db, 
             dbtable = tbl, 
             driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"), 
        options), 
      repartition = repartition, ...)
}

我在运行时将repartition设置为24。因此，我看不出与建议的帖子有什么联系。

编辑2
我可以通过摆脱重新分区来解决我的问题。有人能解释为什么在这种情况下使用sparklyr重新分区是无效的吗？

Answer 1

正如在另一个问题中解释的那样，以及其他一些帖子（partitionColumn、lowerBound、upperBound、numPartitions参数的含义是什么？），与从csv文件转换mysql表到spark dataset相比，在spark中通过JDBC读取RDBMS时进行分区，在spark中并行读取mysql数据）和场外资源（并行读取），默认情况下spark JDBC源按顺序将所有数据读取到单个节点中。

有两种并行读取的方式：

>

基于数值列的范围拆分，该列需要lowerbound、upperbound、PartitionColumn和NumPartitions选项，其中PartitionColumn是稳定的数值列（伪olumns可能不是好的选择）

spark_read_jdbc(
  ...,
  options = list(
    ...
    lowerBound = "0",                 # Adjust to fit your data 
    upperBound = "5000",              # Adjust to fit your data 
    numPartitions = "42",             # Adjust to fit your data and resources
    partitionColumn = "some_numeric_column"
  )
 )

谓词列表-sparklyr目前不支持。

重新分区(sparklyr::sdf_repartition)不能解决这个问题，因为它发生在数据加载之后。由于shuffle(repartition)属于Spark中开销最大的操作，它很容易使节点崩溃。

因此使用：

也就是说，如果数据已经由单个节点处理，那么就会产生一个问题，即使用Apache Spark是否有意义。答案将取决于您的管道的其余部分，但仅考虑到有问题的组件，它可能是否定的。