Spark操作parquet文件

package code.parquet

import java.net.URI

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{Path, FileSystem}
import org.apache.spark.sql.{SaveMode, SparkSession}

/**
  * Created by zhen on 2018/12/11.
  */
object ParquetIO {
  // 指定hdfs根节点
  private val hdfsRoot = "hdfs://172.20.32.163:8020"
  // 获取HDFS路径
  def getPath(path: String): Path = {
    if (path.toLowerCase().startsWith("hdfs://")) {
      new Path(path)
    } else {
      new Path(hdfsRoot + path)
    }
  }
  def main(args: Array[String]) {
    val spark = SparkSession.builder().appName("parquet").master("local[2]").getOrCreate()
    spark.sparkContext.setLogLevel("WARN") // 设置日志级别为WARN
    val fsUri = new URI(hdfsRoot)
    val fs = FileSystem.get(fsUri, new Configuration())
    val path = hdfsRoot + "/YXFK/compute/KH_JLD"
    val has = fs.exists(getPath(path))
    if(has){
      // 读取hdfs文件系统parquet数据
      val dataFrame = spark.read.parquet(path)
      dataFrame.show(10)
      // 筛选，过滤数据
      val result = dataFrame.select("JLDBH", "JLDDZ", "JLDMC", "JLFSDM", "CJSJ")
        .filter("JLDDZ is not null AND JLFSDM = 3")
        .sort("JLDBH")
      result.show(10)
      // 写入部分数据到本地
      result.write.mode(SaveMode.Overwrite).parquet("E:\result")
    }
    // 读取本地parquet数据
    val localDataFrame = spark.read.parquet("E:\jld.parquet")
    localDataFrame.show(10)
    // 读取写入数据验证
    val resultSpace = spark.read.parquet("E:\result")
    resultSpace.show(10)
  }
}

结果：

分析：Spark读取parquet数据默认为目录，因此可以只指定到你要读取的上级目录即可（本地模式除外），当保存为parquet时，会自动拆分，因此只能指定为上级目录。