我有一个非常简单的Sparksql连接到Postgres数据库的设置,我试图从表中获取一个DataFrame,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的文档(Spark 1.4.0 – spark-sql_2.11)中,它表示它们用于定义步幅,而不是用于过滤/范围分区列.但这提出了几个问题：
>步幅是每个执行者(分区)查询数据库的频率(每个查询返回的元素数)？
>如果没有,这些参数的目的是什么,它们依赖于什么？如何以稳定的方式平衡我的DataFrame分区(不要求所有分区包含相同数量的元素,只是存在均衡 – 例如2个分区100个元素55 / 45,60 / 40甚至65/35会这样做)
似乎无法找到这些问题的明确答案,并且想知道是否有些人可以为我清除这一点,因为现在正在影响我的集群性能,当处理X万行并且所有繁重的工作都归结为一个遗嘱执行人.
干杯谢谢你的时间.
解决方法
本质上,下限和上限以及分区数用于计算每个并行任务的增量或拆分.
假设该表具有分区列“年”,并且具有2006年至2016年的数据.
如果将分区数定义为10,使用下限2006和更高的绑定2016,则每个任务都会获取其自己年份的数据 – 这是理想情况.
即使您错误地指定了下限和/或上限,例如设置lower = 0和upper = 2016,数据传输会出现偏差,但是,您不会“丢失”或无法检索任何数据,因为：
第一个任务将获取年份的数据< 0. 第二项任务将获取0到2016/10之间的年度数据. 第三项任务将获取2016/10和2 * 2016/10之间的年度数据. … 并且最后一项任务将具有年 – > 2016的where条件. T.
