Spark-SQL的Java实践案例(三)

史上最简单的spark教程
所有代码示例地址:https://github.com/Mydreamandreality/sparkResearch

(提前声明:文章由作者:张耀峰 结合自己生产中的使用经验整理,最终形成简单易懂的文章,写作不易,转载请注明)
(文章参考:Elasticsearch权威指南,Spark快速大数据分析文档,Elasticsearch官方文档,实际项目中的应用场景)
(帮到到您请点点关注,文章持续更新中!)
Git主页 https://github.com/Mydreamandreality

在这里我们继续上一章节SparkSQL的案例啊,没有看上一章节的兄弟萌 点击这里哈https://blog.csdn.net/youbitch1/article/details/88852644

---

这章我们来搞RDD和dataframe的转换操作

---

第一种方式

• Spark-SQL支持两种不同的方式把RDD转换成dataframe
• 第一种方法是根据RDD对象的具体类型映射,推导出schema,也就是反射
• 这种的映射方式代码非常简洁
• 不过前提就是在写application的时候就需要知道schema

第二种方式

• 通过显式的程序代码构造schema,然后将schema应用到RDD上,最后转换成DataSet
• 虽然这种方法编码复杂一些,但是这样我们就可以在不知道DataSet的列名称的类型的情况下使用,在代码运行中读取列和类型

---

简单了解之后我们直接上代码,用代码讲解懂得快

第一种方式:反射的方式推导出schema:代码案例

        SparkSession sparkSession = SparkSession.builder().master("local").appName("Java Spark SQL").getOrCreate();//定义一个RDD,转换成DataFrameJavaRDD<Person> personJavaRDD = sparkSession.read().textFile("URL").javaRDD().map(new Function<String, Person>() {@Overridepublic Person call(String v1) {String[] param = v1.split(":");Person person = new Person();person.setName(param[0]);person.setAge(Integer.valueOf(param[1].trim()));return person;}});//转换成DataFrameDataset<Row> personDataset = sparkSession.createDataFrame(personJavaRDD,Person.class);

代码模拟debug:(给大家解释一波代码)

• 首先我们定义RDD
• 其次map()函数把RDD转换成Person对象
• 最后把Person的RDD对象转换成DataFrame
• 就4这么简单…
• 转换成dataframe之后怎么验证呢,很简单:

代码案例

         //创建临时视图personDataset.createOrReplaceTempView("user");Dataset<Row> result = sparkSession.sql("SELECT * FROM user");Encoder<String> encoder = Encoders.STRING();//第一种方式:通过下标获取valueDataset<String> dataset = result.map(new MapFunction<Row, String>() {@Overridepublic String call(Row value) {return value.getString(0);}},encoder);dataset.show();//第二种方式:通过字段获取valueDataset<String> fieldValue = result.map(new MapFunction<Row, String>() {@Overridepublic String call(Row value) {return value.getAs("name");}},encoder);fieldValue.show();

• 上面的代码就是创建一个临时视图,执行SQL,第一种通过下标的方式获取第一条数据,第二种通过字段名称获取value

---

第二种方式:代码显式的构造schema:代码案例

        SparkSession sparkSession = SparkSession.builder().master("local").appName("spark app").getOrCreate();//创建普通的JavaRDDJavaRDD<String> javaRDD = sparkSession.sparkContext().textFile("URL", 1).toJavaRDD();//字符串编码的模式String schema = "name age";//根据模式的字符串生成模式List<StructField> structFieldList = new ArrayList<>();for (String fieldName : schema.split(" ")) {StructField structField = DataTypes.createStructField(fieldName, DataTypes.StringType, true);structFieldList.add(structField);}StructType structType = DataTypes.createStructType(structFieldList);JavaRDD<Row> rowJavaRDD = javaRDD.map(new Function<String, Row>() {@Overridepublic Row call(String v1) {String[] attirbutes = v1.split(",");return RowFactory.create(attirbutes[0], attirbutes[1].trim());}});//将模式应用于RDDDataset<Row> dataset = sparkSession.createDataFrame(rowJavaRDD, structType);//创建临时视图dataset.createOrReplaceTempView("user");Dataset<Row> result = sparkSession.sql("select * from user");result.show();

• 代码案例解释:

• 首先我们创建JavaRDD
• 其次创建StructType的schema,与rowJavaRDD结构匹配
• 最后通过sparkSession的createDataFrame把schema应用到rowJavaRDD上形成新的dataframe

兄弟萌,RDD转dataframe的核心操作概念以及案例就完事