Fork me on GitHub

Spark与Apache-Parquet

七十年代时,有一长辈连练铁砂掌,功夫成了之后,可以掌断五砖,凌空碎砖,威风得不得了。时至八十年代,只能掌断三砖。到九十年代只能一砖一砖的断了。他说,一直以为功力退步了,后来才知道烧砖的配方改了。

数据压缩

前言

  前两篇将了spark的部署和一些简单的实例Spark初体验(步骤超详细)Spark再体验之springboot整合spark。我相信前两篇会对刚入门的sparker来说会有一些启发。
  今天在使用spark去加载200万条数据的时候,服务器提示内存分配不足(服务器配置较低),这里我就在想有没有什么方法将数据压缩压缩再压缩呢?网上查资料,问他人,最后看到并使用了Apache Parquet官网,这里简单的介绍一下parquet

Parquet

  Parquet是面向分析型业务的列式存储格式,由Twitter和Cloudera合作开发,2015年5月从Apache的孵化器里毕业成为Apache顶级项目.
  当时Twitter的日增数据量达到压缩之后的100TB+,存储在HDFS上,工程师会使用多种计算框架(例如MapReduce, Hive, Pig等)对这些数据做分析和挖掘;日志结构是复杂的嵌套数据类型,例如一个典型的日志的schema有87列,嵌套了7层。所以需要设计一种列式存储格式,既能支持关系型数据(简单数据类型),又能支持复杂的嵌套类型的数据,同时能够适配多种数据处理框架。

通过阅读这篇文章,相信你会对parquet的优点有所了解。
网上关于它的介绍和算法的讲解一大推,简单的讲使用parquet存储数据有以下优点:

  • 压缩数据
  • 不失真
  • 减少IO吞吐量
  • 高效的查询

更重要的是spark和parquet简直是绝配,犹如小葱拌豆腐、京酱肉丝陪面皮、手擀面配大蒜。。。

示例

老规矩,先上pom

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
...
<parquet.version>1.7.0</parquet.version>
...
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-common</artifactId>
<version>${parquet.version}</version>
</dependency>
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-encoding</artifactId>
<version>${parquet.version}</version>
</dependency>
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-column</artifactId>
<version>${parquet.version}</version>
</dependency>
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-hadoop</artifactId>
<version>${parquet.version}</version>
</dependency>

下面使用spark对mysql的一张表做parquet的读写操作
将整张表的数据存储为parquet格式的文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
static String url = "jdbc:mysql://localhost:3306/world?" +
"useUnicode=true&characterEncoding=UTF-8" +
"&zeroDateTimeBehavior=convertToNull";
static String table = "jd_trainingdata";
static String username = "root";
static String passwd = "root";

private static void parquetWrite() {
SparkConf conf = new SparkConf()
.setAppName("test")
.setMaster("local")
.set("spark.executor.memory", "8g")
.set("spark.rdd.compress true","true")
.set("spark.testing.memory", "2147480000");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
sqlContext.setConf("spark.sql.dialect", "sql");
Properties connectionProperties = new Properties();
connectionProperties.put("driver","com.mysql.jdbc.Driver");
connectionProperties.put("user", username);
connectionProperties.put("password", passwd);
DataFrame vectorTable = sqlContext.read().jdbc(url, table,connectionProperties);
vectorTable.saveAsParquetFile("jdData");// 在项目里生成文件,当然你也可以写绝对路径
}

结果为
parquet包含的文件
关于内容我没有去深究(官网有详细的解释说明),但是能看得出文件是成功的生成了~ 那么接下来看看如何去读parquet的文件内容

1
2
3
4
5
6
7
8
9
10
11
12
13
private static void parquetRead() {
SparkConf conf = new SparkConf()
.setAppName("test")
.setMaster("local[4]")
.set("spark.executor.memory", "8g")
.set("spark.rdd.compress true","true")
.set("spark.testing.memory", "2147480000");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
sqlContext.parquetFile("jdData").registerTempTable("jdData");
DataFrame dbaClos = sqlContext.sql("select * from jdData where Title = 'DBA' and Title <> '' ");
dbaClos.show();
}

运行日志
日志
当你拿到sqlContext的时候,你就可以使用spark提供的sql去查找你想要的数据,要提醒一下, spark支持的sql跟mysql,oracle的不太一样,详细参考官网帮助文档

后续

  这里说一下我为什么使用parquet,我现在测试服务器的内存是16g,项目在启动的时候需要加载其他的模型、数据等文件,而且数据虽然才有220多万条,但是存储的内容比较多,所以就会导致内存不足,使用了parquet之后,数据大小直接被压缩为原来的三分之一!!!而且spark对parquet文件的支持近乎完美,所以使用parquet之后,我完全可以不用考虑内存分配不足的问题了。

-------------本文结束感谢您的阅读-------------