SparkSQL：如何在从数据库加载数据集时指定分区列答案

【问题标题】：SparkSQL : How to specify partitioning column while loading dataset from databaseSparkSQL：如何在从数据库加载数据集时指定分区列
【发布时间】：2018-12-26 06:10:18
【问题描述】：

我正在使用 Spark 2.3，并使用 jdbc 从 MySQL 加载数据，如下所示

  val dataSet:Dataset[Row] = _spark
    .read
    .format("jdbc")
    .options(Map("url" -> jdbcUrl
                ,"user" -> username
                ,"password" -> password
                ,"dbtable" -> dataSourceTableName
                ,"driver" -> driver
                ))
    .load()

我想根据表中的特定列对数据集进行分区。我怎样才能做到这一点？

【问题讨论】：

标签： apache-spark apache-spark-sql

【解决方案1】：

您需要指定partitionColumn、upperBound、lowerBound 和numPartitions 选项。

这些在JDBC documentation for spark sql 的属性表中进行了描述。

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

有关upperBound 和lowerBound 参数的进一步说明，请参见@PIYUSH PASARI 的answer。

他给出了以下使用以下参数值生成的查询示例

upperBound = 500、lowerBound = 0 和 numPartitions = 5。

SELECT * FROM table WHERE partitionColumn < 100 or partitionColumn is null
SELECT * FROM table WHERE partitionColumn >= 100 AND <200 
SELECT * FROM table WHERE partitionColumn >= 200 AND <300
SELECT * FROM table WHERE partitionColumn >= 300 AND <400
...
SELECT * FROM table WHERE partitionColumn >= 400

这可以从JDBCRelation.scala中的代码看出。

如您所见，所有行都已获取，但如果您的上限和下限未覆盖整个数据范围，则第一个和最后一个分区可能比其他分区大。如果您不能确定上限和下限，想要偶数分区并且不关心获取每一行，您可以始终将上限和下限设置为 dbtable 参数中的条件。

【讨论】：

谢谢，有没有办法避免给出“upperBound and lowerBound”而只给出partitionColumn
不必指定每个参数。它们仅用于指定每个分区的步长。将返回所有行，但如果您的边界不正确，则第一个和最后一个分区可能大于其余分区。如果想要偶数分区并且只想要给定范围内的结果，您还应该在查询中设置上限和下限。

【解决方案2】：

spark.read("jdbc")
  .option("url", url)
  .option("dbtable", "pets")
  .option("user", user)
  .option("password", password)
  .option("numPartitions", 10)
  .option("partitionColumn", "owner_id")
  .option("lowerBound", 1)
  .option("upperBound", 10000)

阅读以下链接了解更多

【讨论】：