Hey there! Let's tackle your problem head-on: when you read data from Hive into a DataFrame and convert it to a JavaRDD, all subsequent transformations run on a single task. Using repartition fixes the parallelism but brings expensive shuffles—totally understand why you want a better way.

First, why is this happening? #

The root cause is almost always that your initial DataFrame has only 1 partition. This happens if your Hive table has too few files (e.g., one huge file) or Spark can't properly split the data during reading. When you convert this to a JavaRDD, it inherits that single partition, so all map/flatMap operations are stuck on one task.

Better alternatives to repartition (no heavy shuffles) #

1. Fix the source Hive table (most optimal)

The best long-term fix is to ensure your Hive table is structured to let Spark create enough partitions automatically:

• For columnar formats (Parquet/ORC):
  • If you have too few large files, use ALTER TABLE my_table CONCATENATE to merge small files (or split large ones by rewriting the table). This ensures Spark reads each file as a separate partition.
  • When writing to the Hive table (e.g., via INSERT OVERWRITE), set spark.sql.shuffle.partitions to a reasonable value (matching your cluster's parallelism) to generate multiple output files.
• For text-based formats:
  • Adjust spark.sql.files.maxPartitionBytes (default 128MB) to a smaller value (e.g., 64MB) so Spark splits large files into more partitions. You can set this before reading the table:

    df.sqlContext().conf().set("spark.sql.files.maxPartitionBytes", "67108864"); // 64MB
    

2. Adjust partitioning during DataFrame reading (no shuffle, or minimal shuffle)

Instead of shuffling later with repartition, tweak how Spark reads the data:

• Use partition-aware reading:
  If your Hive table is partitioned by a column (e.g., date), make sure your query filters on that partition. Spark will automatically create partitions for each Hive partition, avoiding manual repartitioning.
• Repartition based on a business field (lower shuffle cost):
  If you must repartition, do it on a field you'll use later (like your dimensionName). This keeps related data together and reduces shuffle overhead compared to random repartitioning:

  DataFrame tempDf = df.sqlContext().sql("SELECT * FROM my_table");
  // Repartition using your dimension field instead of a random number
  DataFrame partitionedDf = tempDf.repartition(col(dimensionName));
  JavaRDD<IMSSummaryPOJO> inputData = partitionedDf.toJavaRDD().flatMap(...);
  

3. Check Spark's Hive compatibility settings

Ensure Spark is properly interpreting your Hive table's metadata:

• Enable spark.sql.hive.convertMetastoreParquet (default true) to let Spark directly read Parquet files instead of going through Hive's SerDe, which helps it correctly split files into partitions.
• For older Hive tables, you might need to refresh the table metadata in Spark with df.sqlContext().sql("REFRESH TABLE my_table");

Quick code adjustment example #

Here's how you could modify your code to use the optimal approach (fixing source + reading with proper partitions):

// First, adjust Spark config to split files into smaller partitions
df.sqlContext().conf().set("spark.sql.files.maxPartitionBytes", "67108864");

// Read from Hive—now Spark will create multiple partitions based on file size
DataFrame tempDf = df.sqlContext().sql("SELECT * FROM my_table");

// Verify partition count (optional, for debugging)
// System.out.println("Number of partitions: " + tempDf.rdd().getNumPartitions());

// Convert to RDD and run transformations—no need for repartition!
JavaRDD<IMSSummaryPOJO> inputData = tempDf.toJavaRDD().flatMap(new FlatMapFunction<Row, IMSSummaryPOJO>() {
    @Override
    public Iterator<IMSSummaryPOJO> call(Row row) throws Exception {
        // Your flatMap logic here
    }
});

JavaPairRDD<Text,IMSMetric> inputRecordRdd = inputData.flatMapToPair(new IMSInputRecordFormat(dimensionName,hllCounterPValue,hllCounterKValue,dimensionConfigMapBroadCast));

Final notes #

• Always check the number of partitions in your initial DataFrame first (tempDf.rdd().getNumPartitions())—this tells you if the issue is with reading or later transformations.
• Avoid repartition unless absolutely necessary—shuffling data across the cluster is one of the most expensive operations in Spark.

内容的提问来源于stack exchange，提问作者Makubex