spark jdbc df limit ...它在做什么？

Question

我正在尝试学习如何了解 Spark 内部正在发生的事情，这是我目前的困惑。我正在尝试将 Oracle 表中的前 200 行读入 Spark：

val jdbcDF = spark.read.format("jdbc").options(
  Map("url" -> "jdbc:oracle:thin:...",
  "dbtable" -> "schema.table",
  "fetchSize" -> "5000",
  "partitionColumn" -> "my_row_id",
  "numPartitions" -> "16",
  "lowerBound" -> "0",
  "upperBound" -> "9999999"
  )).load()

jdbcDF.limit(200).count()

我希望这会相当快。对具有 500K 行的表的类似操作在合理的时间内完成。在这种特殊情况下，表要大得多（数亿行），但我认为 limit(200) 会使其更快吗？我该如何弄清楚它把时间花在了哪里？

Answer 1

事实上，spark 还不能下推limit 谓词。

所以实际上在这种情况下发生的情况是它正在拉动所有数据以激发火花，然后进行限制和计数。您需要在子查询中将其用作表参数。

例如：

val jdbcDF = spark.read.format("jdbc").options(
  Map("url" -> "jdbc:oracle:thin:...",
  "dbtable" -> "(select * from schema.table limit 200) as t",
  "fetchSize" -> "5000",
  "partitionColumn" -> "my_row_id",
  "numPartitions" -> "16",
  "lowerBound" -> "0",
  "upperBound" -> "9999999"
  )).load()

因此，主要花费时间的地方是提取所有数据以激发火花。

也可以在子查询中动态传递限制：

val n : Int = ???

val jdbcDF = spark.read.format("jdbc").options(
  Map("url" -> "jdbc:oracle:thin:...",
  "dbtable" -> s"(select * from schema.table limit $n) as t",
  "fetchSize" -> "5000",
  "partitionColumn" -> "my_row_id",
  "numPartitions" -> "16",
  "lowerBound" -> "0",
  "upperBound" -> "9999999"
  )).load()

有一个JIRA ticket (SPARK-10899)正在进行来解决这个问题，但它已经挂了将近一年了。

编辑：因为上述 JIRA 中的问题被标记为重复。您可以继续跟踪问题here - SPARK-12126。 我希望这能回答你的问题。