在不知道上限的情况下使用 spark 从 sql 数据库中并行读取

Question

Spark 允许您从 sql db 源并行读取，并且可以基于滑动窗口进行分区，例如（来自book，第 7 章）

 val colName = "count"
 val lowerBound = 0L
 val upperBound = 348113L // this is the max count in our table
 val numPartitions = 10

 spark.read.jdbc(url,
                tablename,
                colName,
                lowerBound,
                upperBound,
                numPartitions,
                props).count()

在这里，上限是事先知道的。

比方说，一张表在一天内获得了“x”行（可能在 1-2 百万之间），在一天结束时，我们提交了一个 spark 作业，进行一些转换并写入 Parquet/ CSV/JSON。如果我们事先不知道将有多少行（从 1 到 2 百万不等）写入 SQL 源数据库，那么在这种情况下，进行分区的最佳方法或做法是什么。

一种方法是估计您的上限，但我不确定这是一种正确的方法。

Answer 1

我遇到了完全相同的问题。我也需要随机分布。所以我选择一个 int 列并在其上获取 mod 10。这样我就可以在不关心的情况下分割任何东西 界限或分布。

options += ("numPartitions" -> numPartitions,"partitionColumn"->"mod(my_int_column,10)","lowerBound"->"0","upperBound"->"9")

Answer 2

每天的行数都不同这一事实并没有太大变化。假设您想要有 20 个分区，那么有一天您在单个分区中大约有 1M/20 行，而另一天大约有 2M/20。如果有更多的数据并且分区的数量是固定的，那么显然分区中会有更多的数据。

如果对下限和上限有混淆，我想澄清一下，lowerBound 和 upperBound 指的是您要分区的列的值，而不是行数。此外，表格不会根据这些值进行过滤，这意味着如果您的行的值小于 lowerBound 或大于 upperBound，这些行仍将被包括在内。

见： Whats meaning of partitionColumn, lowerBound, upperBound, numPartitions parameters?

来自文档：http://spark.apache.org/docs/latest/sql-programming-guide.html#jdbc-to-other-databases

partitionColumn、lowerBound、upperBound：如果指定了这些选项中的任何一个，则必须全部指定。此外，必须指定 numPartitions。他们描述了从多个工作人员并行读取时如何对表进行分区。 partitionColumn 必须是相关表中的数字列。请注意，lowerBound 和 upperBound 仅用于决定分区步长，而不是用于过滤表中的行。所以表中的所有行都会被分区并返回。此选项仅适用于阅读。