七十年代时，有一长辈连练铁砂掌，功夫成了之后，可以掌断五砖，凌空碎砖，威风得不得了。时至八十年代，只能掌断三砖。到九十年代只能一砖一砖的断了。他说，一直以为功力退步了，后来才知道烧砖的配方改了。
前言
前两篇将了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
...
<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格式的文件
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的文件内容
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之后，我完全可以不用考虑内存分配不足的问题了。
