Parquet介绍及简单使用

桃扇骨 2022-10-02 10:58 278阅读 0赞

> ### **什么是parquet** ###

Parquet 是列式存储的一种文件类型

官网描述：

Apache Parquet is a columnar storage format available to any project in the Hadoop ecosystem, regardless of the choice of data processing framework, data model or programming language

无论数据处理框架，数据模型或编程语言的选择如何，Apache Parquet都是Hadoop生态系统中任何项目可用的列式存储格式

> ### **由来** ###

Parquet的灵感来自于2010年Google发表的Dremel论文，文中介绍了一种支持嵌套结构的存储格式，并且使用了列式存储的方式提升查询性能，在Dremel论文中还介绍了Google如何使用这种存储格式实现并行查询的，如果对此感兴趣可以参考论文和开源实现Apache Drill。

> ### **特点** ###

*  **可以跳过不符合条件的数据**，只读取需要的数据，降低 IO 数据量
 *  **压缩编码可以降低磁盘存储空间**（由于同一列的数据类型是一样的，可以使用更高效的压缩编码（如 Run Length Encoding t  Delta Encoding）进一步节约存储空间）
 *  **只读取需要的列**，支持向量运算，能够获取更好的扫描性能
 *  **Parquet 格式是 Spark SQL 的默认数据源**，可通过 spark.sql.sources.default 配置

> ### **parquet 常用操作** ###

### **load 和 save 函数** ###

// 读取 Parquet 文件
    val usersDF = spark.read.load("/test/users.parquet")
    
    // 查询 Schema 和数据
    usersDF.printSchema
    usersDF.show
    
    // 查询用户的 name 和喜爱颜色并保存
    usersDF.select($"name", $"favorite_color").write.save("/test/result/parquet")
    // 验证结果 可通过 printSchema 查询数据结构，使用 show 查看数据
    
    // 显式指定文件格式： 加载 json 格式
    val usersDF = spark.read.format("json").load("/test/people.json")
    
    // 存储模式(Save Modes) 
    // 可以采用 SaveMode 执行存储操作， SaveMode 定义 了对数据的处理模式，需要注意的是，这些保存模式不使用任何锁定，不是原子操作
    // 当使用 Overwrite 方式执行时，在输出新数据之前，原数据就已经被删除
    usersDF.select($"name").write.save("/test/parquet1")   // 若 /test/parquet1 存在会报错
    usersDF.select($"name").wirte.mode("overwrite").save("/test/parquet1")        // 使用 overwrite 即可
    
    // 将结果保存为表, 也可以进行分区， 分桶等操作： partitionBy  bucketBy
    usersDF.select($"name").write.saveAsTable("table1")

### **Parquet文件 ** ###

Parquet 是一个列格式而且用于多个数据处理系统中

Spark SQL 提供支持对于 Parquet 文件的读写，也就是自动保存原始 数据的 Schema, 当写 Parquet 文件时，所有的列被自动转化为 nullable，因为兼容性的缘故

### **读取 Json 格式的数据，将其转换成 parquet 格式，创建相应的表，使用 SQL 语句查询** ###

// 从 json 文件中读入数据
    val empJson = spark.read.json("/test/emp.json")
    // 将数据保存为 parquet
    empJson.write.mode("overwrite").parquet("/test/parquet")
    // 读取 parquet
    val empParquet = spark.read.parquet("/test/parquet")
    // 创建临时表 emptable
    empParquet.createOrReplaceTempView("emptalbe")
    // 使用 SQL 语句执行查询
    spark.sql("select * from emptable where deptno=10 and sal>1500").show

### **Schematic 的合并： ** ###

**先定义一个简单的 Schema，然后逐渐增加列描述，用户可以获取多个有多个不同 Schema 但相互兼容的 Parquet 文件**

// 创建第一个文件
    val df1 = sc.makeRDD(1 to 5).map(x=> (x, x*2)).toDF("single", "double")
    scala> df1.printSchema
    root
     |-- single: integer (nullable = false)
     |-- double: integer (nullable = false)
     
     
    // 创建第二个文件 
     scala> val df2 = sc.makeRDD(6 to 10).map(x=> (x, x*2)).toDF("single", "triple")
    df2: org.apache.spark.sql.DataFrame = [single: int, triple: int]
    
    scala> df2.printSchema
    root
     |-- single: integer (nullable = false)
     |-- triple: integer (nullable = false)
      
     scala> df2.write.parquet("/data/testtable/key=2")
    
     // 合并上面的两个文件
    scala> val df3 = spark.read.option("mergeSchema", "true").parquet("/data/testtable")
    df3: org.apache.spark.sql.DataFrame = [single: int, double: int ... 2 more fields]
    
    scala> df3.printSchema
    root
     |-- single: integer (nullable = true)
     |-- double: integer (nullable = true)
     |-- triple: integer (nullable = true)
     |-- key: integer (nullable = true)
     
     scala> df3.show
    +------+------+------+---+
    |single|double|triple|key|
    +------+------+------+---+
    |     8|  null|    16|  2|
    |     9|  null|    18|  2|
    |    10|  null|    20|  2|
    |     3|     6|  null|  1|
    |     4|     8|  null|  1|
    |     5|    10|  null|  1|
    |     6|  null|    12|  2|
    |     7|  null|    14|  2|
    |     1|     2|  null|  1|
    |     2|     4|  null|  1|
    +------+------+------+---+

### ** Json Datasets（两种写法）** ###

// 第一种
    scala> val df4 = spark.read.json("/app/spark-2.2.1-bin-hadoop2.7/examples/src/main/resources/people.json")
    df4: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
    
    scala> df4.show
    +----+-------+
    | age|   name|
    +----+-------+
    |null|Michael|
    |  30|   Andy|
    |  19| Justin|
    +----+-------+
    
    // 第二种
    scala> val df5 = spark.read.format("json").load("/app/spark-2.2.1-bin-hadoop2.7/examples/src/main/resources/people.json")
    df5: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
    
    scala> df5.show
    +----+-------+
    | age|   name|
    +----+-------+
    |null|Michael|
    |  30|   Andy|
    |  19| Justin|
    +----+-------+

### **JDBC 方式读取关系型数据库中的数据（需要将 JDBC 的驱动加入）** ###

// 将 JDBC 的驱动加入
    bin/spark-shell --master spark://bigdata11:7077 --jars /root/temp/ojdbc6.jar --driver-class-path /root/temp/ojdbc6.jar
    
    // 读取 Oracle
    val oracleEmp = spark.read.format("jdbc")
                        .option("url","jdbc:oracle:thin:@192.168.10.100:1521/orcl.example.com")
                        .option("dbtable","scott.emp")
                        .option("user","scott")
                        .option("password","tiger").load

### ** 操作 Hive 的表** ###

**把 hive 和 hadoop 的配置文件拷贝到sprke 的 conf 目录下： hive-sit.xml， core-sit.xml, hdfs-sit.xml**

**启动 Spark-shell 时 指定mysql 数据库的驱动程序**

./bin/spark-shell --master spark://bigdata0:7077 --jars /data/tools/mysql-connector-java-5.1.43-bin.jar  --driver-class-path /data/tools/mysql-connector-java-5.1.43-bin.jar

** 使用 Spark Shell 操作 Hive**

// 创建表
    spark.sql("create table ccc(key INT, value STRING) row format delimited fields terminated by ','")
    
    // 导入数据
    spark.sql("load data local path '/test/data.txt' into table ccc")
    
    // 查询数据
    spark.sql("select * from ccc").show

**使用 Spark SQL 操作 Hive**

show tables;
    select * from ccc;

转：[https://blog.51cto.com/songqinglong/2084049][https_blog.51cto.com_songqinglong_2084049]

[https_blog.51cto.com_songqinglong_2084049]: https://blog.51cto.com/songqinglong/2084049