将 Dataframe 转换为 RDD 减少了分区

Question

在我们的代码中，Dataframe 被创建为：

DataFrame DF = hiveContext.sql("select * from table_instance");

当我将数据帧转换为 rdd 并尝试获取其分区数时

RDD<Row> newRDD = Df.rdd();
System.out.println(newRDD.getNumPartitions());

它将分区的数量减少到 1（1 打印在控制台中）。最初我的数据框有 102 个分区。

更新：

在阅读时，我重新分配了数据框：

DataFrame DF = hiveContext.sql("select * from table_instance").repartition(200);

然后转换为 rdd ，所以它只给了我 200 个分区。 有没有

JavaSparkContext

在这方面有什么作用吗？当我们将数据帧转换为 rdd 时，默认最小分区标志是否也在 spark 上下文级别考虑？

更新：

我制作了一个单独的示例程序，其中我将完全相同的表读入数据帧并转换为 rdd。没有为 RDD 转换创建额外的阶段，并且分区计数也是正确的。我现在想知道我在主程序中做了什么不同。

如果我的理解有误，请告诉我。

Answer 1

它基本上取决于hiveContext.sql()的实现。由于我是 Hive 新手，我的猜测是 hiveContext.sql 不知道 OR 无法拆分表中存在的数据。

例如，当您从 HDFS 读取文本文件时，spark 上下文会考虑该文件使用的块数来确定分区。

您对repartition 所做的操作是解决此类问题的明显方法。（注意：如果不使用适当的分区器，重新分区可能会导致随机操作，默认使用哈希分区器）

引起您的怀疑，hiveContext 可能会考虑默认的最小分区属性。但是，依赖默认属性不会 解决你所有的问题。例如，如果您的 hive 表的大小增加，您的程序仍然使用默认的分区数。

更新：重新分区期间避免随机播放

定义您的自定义分区器：

public class MyPartitioner extends HashPartitioner {
    private final int partitions;
    public MyPartitioner(int partitions) {
        super();
        this.partitions = partitions;
    }
    @Override
    public int numPartitions() {
        return this.partitions;
    }

    @Override
    public int getPartition(Object key) {
        if (key instanceof String) {
            return super.getPartition(key);
        } else if (key instanceof Integer) {
            return (Integer.valueOf(key.toString()) % this.partitions);
        } else if (key instanceof Long) {
            return (int)(Long.valueOf(key.toString()) % this.partitions);
        }
        //TOD ... add more types
    }
}

使用您的自定义分区器：

JavaPairRDD<Long, SparkDatoinDoc> pairRdd = hiveContext.sql("select *   from table_instance")
.mapToPair( //TODO ... expose the column as key)

rdd = rdd.partitionBy(new MyPartitioner(200));
//... rest of processing