Spark2.2.0中RDD转DataFrame的方式是怎样的
Spark2.2.0中RDD转DataFrame的方式是怎样的,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。

让客户满意是我们工作的目标,不断超越客户的期望值来自于我们对这个行业的热爱。我们立志把好的技术通过有效、简单的方式提供给客户,将通过不懈努力成为客户在信息化领域值得信任、有价值的长期合作伙伴,公司提供的服务项目有:域名申请、虚拟主机、营销软件、网站建设、贵阳网站维护、网站推广。
Spark SQL将现有的RDDs转换为数据集。
方法:使用反射来推断包含特定对象类型的RDD的模式。这种基于反射的方法使代码更加简洁,并且当您在编写Spark应用程序时已经了解了模式时,它可以很好地工作。
第一种方法代码实例java版本实现:
数据准备studentDatatxt
1001,20,zhangsan1002,17,lisi1003,24,wangwu1004,16,zhaogang
本地模式代码实现:
package com.unicom.ljs.spark220.study;import org.apache.spark.SparkConf;import org.apache.spark.api.java.JavaRDD;import org.apache.spark.api.java.JavaSparkContext;import org.apache.spark.api.java.function.Function;import org.apache.spark.api.java.function.VoidFunction;import org.apache.spark.sql.Dataset;import org.apache.spark.sql.Row;import org.apache.spark.sql.SQLContext;/*** @author: Created By lujisen* @company ChinaUnicom Software JiNan* @date: 2020-01-20 08:58* @version: v1.0* @description: com.unicom.ljs.spark220.study*/public class RDD2DataFrameReflect {public static void main(String[] args) {SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD2DataFrameReflect");JavaSparkContext sc = new JavaSparkContext(sparkConf);SQLContext sqlContext=new SQLContext(sc);JavaRDDlines = sc.textFile("C:\\Users\\Administrator\\Desktop\\studentData.txt"); JavaRDDstudentRDD = lines.map(new Function () { @Overridepublic Student2 call(String line) throws Exception {String[] split = line.split(",");Student2 student=new Student2();student.setId(Integer.valueOf(split[0]));student.setAge(Integer.valueOf(split[1]));student.setName(split[2]);return student;}});//使用反射方式将RDD转换成dataFrame//将Student.calss传递进去,其实就是利用反射的方式来创建DataFrameDatasetdataFrame = sqlContext.createDataFrame(studentRDD, Student2.class);
//拿到DataFrame之后将其注册为临时表,然后针对其中的数据执行SQL语句dataFrame.registerTempTable("studentTable");//针对student临时表,执行sql语句查询年龄小于18岁的学生,/*DataFrame rowDF */Datasetdataset = sqlContext.sql("select * from studentTable where age < 18");
JavaRDDrowJavaRDD = dataset.toJavaRDD();
JavaRDDageRDD = rowJavaRDD.map(new Function () {
@Overridepublic Student2 call(Row row) throws Exception {Student2 student = new Student2();student.setId(row.getInt(0));student.setAge(row.getInt(1));student.setName(row.getString(2));return student;}});ageRDD.foreach(new VoidFunction() { @Overridepublic void call(Student2 student) throws Exception {System.out.println(student.toString());}});}}
Student2类:
package com.unicom.ljs.spark220.study;import java.io.Serializable;/*** @author: Created By lujisen* @company ChinaUnicom Software JiNan* @date: 2020-01-20 08:57* @version: v1.0* @description: com.unicom.ljs.spark220.study*/public class Student2 implements Serializable {int id;int age;String name;public int getId() {return id;}public void setId(int id) {this.id = id;}public int getAge() {return age;}public void setAge(int age) {this.age = age;}public String getName() {return name;}public void setName(String name) {this.name = name;}@Overridepublic String toString() {return "Student2{" +"id=" + id +", age=" + age +", name='" + name + '\'' +'}';}}
pom.xml关键依赖:
2.2.0
2.11.8 org.apache.spark spark-sql_2.11 ${spark.version} org.apache.spark spark-core_2.11 ${spark.version}
看完上述内容,你们掌握Spark2.2.0中RDD转DataFrame的方式是怎样的的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注创新互联行业资讯频道,感谢各位的阅读!
分享文章:Spark2.2.0中RDD转DataFrame的方式是怎样的
浏览地址:http://www.jxjierui.cn/article/gcchco.html


咨询
建站咨询
