SparkSQL PostgresQL数据共享分区

Question

我有一个非常简单的SparkSQL连接到Postgres的设置，我正在尝试从一个表中获得一个DataFrame，这个表是一个具有X个分区(假设为2)的数据。守则如下：

Map<String, String> options = new HashMap<String, String>();
options.put("url", DB_URL);
options.put("driver", POSTGRES_DRIVER);
options.put("dbtable", "select ID, OTHER from TABLE limit 1000");
options.put("partitionColumn", "ID");
options.put("lowerBound", "100");
options.put("upperBound", "500");
options.put("numPartitions","2");
DataFrame housingDataFrame = sqlContext.read().format("jdbc").options(options).load();

由于某种原因，DataFrame的一个分区几乎包含所有行。

对于我所能理解的，lowerBound/upperBound是用来细化这一点的参数。在SparkSQL的文档(Spark1.4.0-Spark-SQL_2.11)中，它说它们用于定义步长，而不是过滤/范围分区列。但这提出了几个问题：

1. 步幅是火花将查询每个执行器(分区)的DB的频率(每个查询返回的元素数)吗？
2. 如果不是，这些参数的目的是什么，它们依赖什么，以及我如何以稳定的方式平衡我的DataFrame分区(不要求所有分区包含相同数量的元素，只是有一个平衡-例如，2个分区100个元素55/45、60/40甚至65/35 )

似乎找不到这些问题的明确答案，我想知道你们中的一些人是否可以帮我澄清这一点，因为现在在处理X百万行时影响到我的集群性能，所有的繁重工作都由一个执行者来完成。

干杯，谢谢你抽出时间。

Answer 1

下界确实是针对分区列使用的；请参阅此代码(编写此代码时的当前版本)：

https://github.com/apache/spark/blob/40ed2af587cedadc6e5249031857a922b3b234ca/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala

函数columnPartition包含用于分区逻辑和使用下限/上限的代码。

Answer 2

本质上，下界和上界以及分区数用于计算每个并行任务的增量或拆分。

假设该表有分区列“年份”，并有2006至2016年的数据。

如果将分区的数量定义为10个，其中包含下限2006和更高的界限2016，那么每个任务都将获取其自己年份的数据--这是理想的情况。

即使您错误地指定了下限和/或上限，例如set下限=0和but = 2016，数据传输也会出现倾斜，但是，您不会“丢失”或无法检索任何数据，因为：

第一个任务将获取年份< 0的数据。

第二项任务将获取0至2016/10年间的数据。

第三项任务将获取2016/10至2*2016/10年度的数据。

..。

最后一项任务将在2016年->2016年完成。

在T。

Answer 3

低界和上界目前已被确定来做他们在以前的答案中所做的事情。后续工作将是如何平衡跨分区的数据，而不查看最小最大值，或者数据是否严重倾斜。

如果您的数据库支持“散列”函数，那么它就可以做到这一点。

partitionColumn =“散列(Column_name)%num_partitions”

numPartitions = 10 //任何你想要的

lowerBound =0

upperBound = numPartitions

只要模数运算在[0，numPartitions]上返回均匀分布，这将起作用。