Spark并行读取Hive分区表emp1写入Parquet表EMP2的实现咨询

一、核心实现代码（Scala示例） #

直接通过Spark对接Hive元数据完成转换，全程无需指定HDFS路径：

import org.apache.spark.sql.SparkSession

object HiveTextToParquet {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("HiveTextToParquet")
      .enableHiveSupport() // 启用Hive元数据对接
      .getOrCreate()

    // 读取emp1表，并控制并行处理的分区数为10
    val empDF = spark.sql("SELECT * FROM emp1")
      .repartition(10) // 强制将数据切分为10个并行任务
      // 若需精准读取指定10个分区，可替换为：
      // .where("partition_col IN (SELECT DISTINCT partition_col FROM emp1 LIMIT 10)")

    // 写入Parquet格式的Hive表EMP2
    empDF.write
      .mode("overwrite") // 根据业务需求选择append/overwrite
      .format("parquet")
      .saveAsTable("EMP2") // 直接保存为Hive管理表，无需手动指定路径

    spark.stop()
  }
}

二、需求对应实现细节 #

1. 并行读取10个分区并写入 #

• 通用方案：使用repartition(10)将数据集重分区为10份，Spark会自动并行处理这10个分区的读取与写入操作，满足“并行处理10个分区”的要求。
• 精准指定分区：若需读取特定的10个分区而非随机，可通过SQLWHERE子句锁定分区列取值，比如WHERE partition_col IN (SELECT DISTINCT partition_col FROM emp1 LIMIT 10)，此时Spark仅读取指定的10个分区，天然形成10个并行任务。

2. 确保多Executor并行处理 #

通过Spark提交参数控制Executor资源，示例提交命令：

spark-submit \
  --class HiveTextToParquet \
  --master yarn \
  --num-executors 5 \ # 指定Executor数量，根据集群资源调整
  --executor-cores 2 \ # 每个Executor的CPU核数
  --executor-memory 4G \ # 每个Executor的内存分配
  --driver-memory 2G \
  your-application.jar

• --num-executors直接指定启动的Executor数量，确保多节点并行执行任务；
• 搭配--executor-cores和--executor-memory优化每个Executor的资源配额，提升单节点处理效率，缩短整体转换时长。

3. 不使用HDFS路径的关键 #

• 全程基于表名操作：通过spark.sql("SELECT * FROM emp1")读取Hive表，saveAsTable("EMP2")直接保存为Hive管理表，所有路径逻辑由Spark和Hive元数据自动维护，无需手动指定任何HDFS路径；
• 必须启用enableHiveSupport()，让Spark能够直接对接Hive元数据服务，实现无路径化的表操作。

三、注意事项 #

• 提前确认EMP2表的结构与emp1匹配，或让Spark自动根据数据生成表结构；
• 若emp1为多分区表，Spark会自动识别分区信息，写入EMP2时会保留分区结构（Parquet格式原生支持分区）；
• 重分区数量（10）可根据单分区数据量调整，避免因分区过大导致OOM，或分区过小造成资源浪费。

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