使用 jdbc 驱动程序读取大表时超时和内存不足错误

Question

我正在尝试在 scala 中使用 spark 的本机 read.jdbc 从 Oracle 数据库中将一个大表读入 spark 数据帧。我已经用中小型表（最多 11M 行）对此进行了测试，它工作得很好。但是，当尝试引入更大的表（约 70M 行）时，我不断收到错误。

显示我如何阅读本文的示例代码：

val df = sparkSession.read.jdbc(
   url = jdbcUrl,
   table = "( SELECT * FROM keyspace.table WHERE EXTRACT(year FROM date_column) BETWEEN 2012 AND 2016)"
      columnName = "id_column", // numeric column, 40% NULL
      lowerBound = 1L,
      upperBound = 100000L,
      numPartitions = 60, // same as number of cores
      connectionProperties = connectionProperties) // this contains login & password

我正在尝试并行化操作，因为我正在使用一个具有 60 个内核和 6 x 32GB RAM 的集群专用于此应用程序。但是，我仍然不断收到与超时和内存不足问题相关的错误，例如：

17/08/16 14:01:18 WARN Executor: Issue communicating with driver in heartbeater
org.apache.spark.rpc.RpcTimeoutException: Futures timed out after [10 seconds]. This timeout is controlled by spark.executor.heartbeatInterval
at org.apache.spark.rpc.RpcTimeout.org$apache$spark$rpc$RpcTimeout$$createRpcTimeoutException(RpcTimeout.scala:47)
....

Caused by: java.util.concurrent.TimeoutException: Futures timed out after [10 seconds} 

...

17/08/16 14:17:14 ERROR RetryingBlockFetcher: Failed to fetch block rdd_2_89, and will not retry (0 retries)
    org.apache.spark.network.client.ChunkFetchFailureException: Failure while fetching StreamChunkId{streamId=398908024000, chunkIndex=0}: java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE
      at sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:869)
      at org.apache.spark.storage.DiskStore$$anonfun$getBytes$4.apply(DiskStore.scala:125)
...


17/08/16 14:17:14 WARN BlockManager: Failed to fetch block after 1 fetch failures. Most recent failure cause:
org.apache.spark.SparkException: Exception thrown in awaitResult: 
at org.apache.spark.util.ThreadUtils$.awaitResult(ThreadUtils.scala:205)

集群中应该有足够多的 RAM 用于这种大小的表（我已经在本地表中读取了 10 倍大），所以我觉得由于某种原因可能不会并行读取数据？查看 spark UI 中的时间线，我可以看到一个执行程序挂起并且正在“计算”很长时间。现在，分区列中有很多 NULL 值（大约 40%），但它是唯一的数字列（其他的是日期和字符串） - 这会有所不同吗？还有其他方法可以并行化 jdbc 读取吗？

Answer 1

分区列中有很多 NULL 值（大约 40%），但它是唯一的数字列（其他的是日期和字符串） - 这会有所不同吗？

这有很大的不同。所有带有NULL will go to the last partition 的值：

val whereClause =
  if (uBound == null) {
    lBound
  } else if (lBound == null) {
    s"$uBound or $column is null"
  } else {
    s"$lBound AND $uBound"
}

还有其他方法可以并行化 jdbc 读取吗？

您可以将predicates 与数字列以外的其他列一起使用。例如，您可以在table 中使用ROWID pseudocoulmn，并使用一系列基于前缀的谓词。