Java中如何使用Spark仅通过SparkContext读取parquet到JavaRDD

完全可以仅通过SparkContext实现Parquet文件读取并生成JavaRDD，不需要依赖SparkSQL相关模块。

核心实现逻辑是借助Parquet官方提供的Hadoop InputFormat适配接口，调用SparkContext原生支持的newAPIHadoopFile方法读取数据源，再提取结果即可。

前置依赖 #

你需要确保项目中引入和Spark版本兼容的parquet-hadoop依赖，不需要引入任何Spark SQL相关依赖包。

代码示例 #

import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.hadoop.ParquetInputFormat;
import org.apache.parquet.hadoop.api.ReadSupport;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.simple.SimpleGroupReadSupport;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import scala.Tuple2;

public class ParquetReader {
    public JavaRDD<Group> readParquetToRDD(SparkContext sc, String parquetFilePath) {
        Configuration hadoopConf = new Configuration();
        // 配置Parquet读取的ReadSupport，通用读取用SimpleGroupReadSupport即可
        hadoopConf.set(ReadSupport.PARQUET_READ_SUPPORT, SimpleGroupReadSupport.class.getName());

        // 调用SparkContext的新Hadoop API读取Parquet文件
        JavaPairRDD<Void, Group> rawPairRDD = sc.newAPIHadoopFile(
                parquetFilePath,
                ParquetInputFormat.class,
                Void.class,
                Group.class,
                hadoopConf
        ).toJavaRDD();

        // 提取PairRDD中的Value部分，得到最终的JavaRDD
        return rawPairRDD.map(Tuple2::_2);
    }
}

注意事项 #

• 如果你需要读取自定义的Parquet序列化实体类，可以将SimpleGroupReadSupport替换为对应实体的SpecificParquetReadSupport，同时将返回值泛型从Group修改为你的自定义类即可
• 所有Parquet原生支持的读取配置（比如列裁剪、行组过滤、编码适配等）都可以直接设置到上述hadoopConf对象中，用法和原生Parquet读取完全一致
• 该实现完全不依赖Spark SQL、Dataset相关模块，仅基于Spark Core和Parquet官方Hadoop适配包实现

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