Java和scala实现 Spark RDD转换成DataFrame的两种方法小结

 更新时间:2018年06月07日 09:04:25   作者:黑白调92  
今天小编就为大家分享一篇Java和scala实现 Spark RDD转换成DataFrame的两种方法小结,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧

一:准备数据源

在项目下新建一个student.txt文件,里面的内容为:

1,zhangsan,20 
2,lisi,21 
3,wanger,19 
4,fangliu,18 

二:实现

Java版:

1.首先新建一个student的Bean对象,实现序列化和toString()方法,具体代码如下:

package com.cxd.sql;
import java.io.Serializable;
@SuppressWarnings("serial")
public class Student implements Serializable {
 String sid;
 String sname;
 int sage;
 public String getSid() {
  return sid;
 }
 public void setSid(String sid) {
  this.sid = sid;
 }
 public String getSname() {
  return sname;
 }
 public void setSname(String sname) {
  this.sname = sname;
 }
 public int getSage() {
  return sage;
 }
 public void setSage(int sage) {
  this.sage = sage;
 }
 @Override
 public String toString() {
  return "Student [sid=" + sid + ", sname=" + sname + ", sage=" + sage + "]";
 }
 
}
		

2.转换,具体代码如下

package com.cxd.sql;
import java.util.ArrayList;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
public class TxtToParquetDemo {
 public static void main(String[] args) {
  
  SparkConf conf = new SparkConf().setAppName("TxtToParquet").setMaster("local");
  SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
  reflectTransform(spark);//Java反射
  dynamicTransform(spark);//动态转换
 }
 
 /**
  * 通过Java反射转换
  * @param spark
  */
 private static void reflectTransform(SparkSession spark)
 {
  JavaRDD<String> source = spark.read().textFile("stuInfo.txt").javaRDD();
  
  JavaRDD<Student> rowRDD = source.map(line -> {
   String parts[] = line.split(",");
   Student stu = new Student();
   stu.setSid(parts[0]);
   stu.setSname(parts[1]);
   stu.setSage(Integer.valueOf(parts[2]));
   return stu;
  });
  
  Dataset<Row> df = spark.createDataFrame(rowRDD, Student.class);
  df.select("sid", "sname", "sage").
  coalesce(1).write().mode(SaveMode.Append).parquet("parquet.res");
 }
 /**
  * 动态转换
  * @param spark
  */
 private static void dynamicTransform(SparkSession spark)
 {
  JavaRDD<String> source = spark.read().textFile("stuInfo.txt").javaRDD();
  
  JavaRDD<Row> rowRDD = source.map( line -> {
   String[] parts = line.split(",");
   String sid = parts[0];
   String sname = parts[1];
   int sage = Integer.parseInt(parts[2]);
   
   return RowFactory.create(
     sid,
     sname,
     sage
     );
  });
  
  ArrayList<StructField> fields = new ArrayList<StructField>();
  StructField field = null;
  field = DataTypes.createStructField("sid", DataTypes.StringType, true);
  fields.add(field);
  field = DataTypes.createStructField("sname", DataTypes.StringType, true);
  fields.add(field);
  field = DataTypes.createStructField("sage", DataTypes.IntegerType, true);
  fields.add(field);
  
  StructType schema = DataTypes.createStructType(fields);
  
  Dataset<Row> df = spark.createDataFrame(rowRDD, schema);
  df.coalesce(1).write().mode(SaveMode.Append).parquet("parquet.res1");
  
  
 }
 
}

scala版本:

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.IntegerType
object RDD2Dataset {
 
 case class Student(id:Int,name:String,age:Int)
 def main(args:Array[String])
 {
 
 val spark=SparkSession.builder().master("local").appName("RDD2Dataset").getOrCreate()
 import spark.implicits._
 reflectCreate(spark)
 dynamicCreate(spark)
 }
 
 /**
	 * 通过Java反射转换
	 * @param spark
	 */
 private def reflectCreate(spark:SparkSession):Unit={
 import spark.implicits._
 val stuRDD=spark.sparkContext.textFile("student2.txt")
 //toDF()为隐式转换
 val stuDf=stuRDD.map(_.split(",")).map(parts⇒Student(parts(0).trim.toInt,parts(1),parts(2).trim.toInt)).toDF()
 //stuDf.select("id","name","age").write.text("result") //对写入文件指定列名
 stuDf.printSchema()
 stuDf.createOrReplaceTempView("student")
 val nameDf=spark.sql("select name from student where age<20")
 //nameDf.write.text("result") //将查询结果写入一个文件
 nameDf.show()
 }
 
 /**
	 * 动态转换
	 * @param spark
	 */
 private def dynamicCreate(spark:SparkSession):Unit={
 val stuRDD=spark.sparkContext.textFile("student.txt")
 import spark.implicits._
 val schemaString="id,name,age"
 val fields=schemaString.split(",").map(fieldName => StructField(fieldName, StringType, nullable = true))
 val schema=StructType(fields)
 val rowRDD=stuRDD.map(_.split(",")).map(parts⇒Row(parts(0),parts(1),parts(2)))
 val stuDf=spark.createDataFrame(rowRDD, schema)
  stuDf.printSchema()
 val tmpView=stuDf.createOrReplaceTempView("student")
 val nameDf=spark.sql("select name from student where age<20")
 //nameDf.write.text("result") //将查询结果写入一个文件
 nameDf.show()
 }
}

注:

1.上面代码全都已经测试通过,测试的环境为spark2.1.0,jdk1.8。

2.此代码不适用于spark2.0以前的版本。

以上这篇Java和scala实现 Spark RDD转换成DataFrame的两种方法小结就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持脚本之家。

相关文章

  • Springboot事务失效的几种情况解读

    Springboot事务失效的几种情况解读

    这篇文章主要介绍了Springboot事务失效的几种情况解读,因为Spring AOP默认使用动态代理,会给被代理的类生成一个代理类,事务相关的操作都通过代理来完成,使用内部方法调用时,使用的是实例调用,没有通过代理类调用方法,因此事务不会检测到失败,需要的朋友可以参考下
    2023-10-10
  • SpringMVC视图作用详解

    SpringMVC视图作用详解

    这篇文章主要介绍了springMVC中的视图与视图解析器,springMVC视图的种类很多,默认有转发视图和重定向视图,本文就每一种视图给大家详细介绍,需要的朋友可以参考下
    2022-11-11
  • SpringBoot如何基于POI-tl和word模板导出庞大的Word文件

    SpringBoot如何基于POI-tl和word模板导出庞大的Word文件

    这篇文章主要介绍了SpringBoot如何基于POI-tl和word模板导出庞大的Word文件,poi-tl是一个基于Apache POI的Word模板引擎,也是一个免费开源的Java类库
    2022-08-08
  • Java实战项目练习之球馆在线预约系统的实现

    Java实战项目练习之球馆在线预约系统的实现

    理论是远远不够的,只有在实战中才能获得能力的提升,本篇文章手把手带你用java+SpringBoot+maven+freemark+Mysql实现一个球馆在线预约系统,大家可以在过程中查缺补漏,提升水平
    2022-01-01
  • 注解@TableName,@TableField,pgsql的模式对应方式

    注解@TableName,@TableField,pgsql的模式对应方式

    这篇文章主要介绍了注解@TableName,@TableField,pgsql的模式对应方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2024-04-04
  • Java关键字finally_动力节点Java学院整理

    Java关键字finally_动力节点Java学院整理

    java关键字finally不管是否出现异常,finally子句总是在块完成之前执行。下面通过实现代码给大家介绍Java关键字finally相关知识,需要的的朋友参考下吧
    2017-04-04
  • SpringCloud Zuul服务功能与使用方法解析

    SpringCloud Zuul服务功能与使用方法解析

    这篇文章主要介绍了SpringCloud Zuul服务功能与使用方法解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-05-05
  • Java开启线程的四种方法案例详解

    Java开启线程的四种方法案例详解

    这篇文章主要介绍了Java开启线程的四种方法,本文结合实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2023-02-02
  • 详谈Lock与synchronized 的区别

    详谈Lock与synchronized 的区别

    下面小编就为大家带来一篇详谈Lock与synchronized 的区别。小编觉得挺不错的,现在就分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2017-05-05
  • springboot @RequiredArgsConstructor的概念与使用方式

    springboot @RequiredArgsConstructor的概念与使用方式

    这篇文章主要介绍了springboot @RequiredArgsConstructor的概念与使用方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2024-09-09

最新评论