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時，會自動拆分，因此只能指定為上級目錄。