Spark中RDD转换成DataFrame的两种方式（分别用Java和scala实现）_spark将personrdd隐式转换为dataframe之后可以用printschema不可用sh

作者：weixin_40725706 | 2024-05-18 16:38:45

踩

spark将personrdd隐式转换为dataframe之后可以用printschema不可用show

转载:http://blog.csdn.net/u010592112/article/details/73730796 感谢原博主分享!!!

一：准备数据源

[java]view plaincopy 
   
   

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

[plain]view plaincopy 
   
 1,zhangsan,20  
 2,lisi,21  
 3,wanger,19  
 4,fangliu,18  

二：实现

Java版：

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

[java]view plaincopy 
   
 import java.io.Serializable;  
   
 public class Student implements Serializable{  
   
     private int id;  
     private String name;  
     private int age;  
     public int getId() {  
         return id;  
     }  
     public void setId(int id) {  
         this.id = id;  
     }  
     public String getName() {  
         return name;  
     }  
     public void setName(String name) {  
         this.name = name;  
     }  
     public int getAge() {  
         return age;  
     }  
     public void setAge(int age) {  
         this.age = age;  
     }  
     @Override  
     public String toString() {  
         return "Student [id=" + id + ", name=" + name + ", age=" + age + "]";  
     }  
       
       
       
 }  

2.转换，具体代码如下

[java]view plaincopy 
   
 import java.util.ArrayList;  
 import java.util.List;  
   
 import org.apache.spark.api.java.JavaRDD;  
 import org.apache.spark.api.java.function.Function;  
 import org.apache.spark.sql.Dataset;  
 import org.apache.spark.sql.Row;  
 import org.apache.spark.sql.RowFactory;  
 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 RDDToDataset {  
   
     public static void main(String[] args) {  
         SparkSession spark=SparkSession.builder()  
                    .appName("RDDToDataset")  
                    .master("local[*]")  
                    .getOrCreate();  
         reflectCreate(spark);  
         dynamicCreate(spark);  
           
     }  
   
     /** 
      * 通过Java反射转换 
      * @param spark 
      */  
     public static void reflectCreate(SparkSession spark)  
     {  
         JavaRDD<Student> stuRDD= spark.read().textFile("student.txt").javaRDD().map(new Function<String, Student>() {  
   
             public Student call(String line) throws Exception {  
               
                 String[] lineSplit=line.split(",");  
                 Student stu=new Student();  
                 stu.setId(Integer.valueOf(lineSplit[0]));  
                 stu.setName(lineSplit[1]);  
                 stu.setAge(Integer.valueOf(lineSplit[2]));  
                 return stu;  
             }  
         });  
           
         Dataset<Row> stuDf=spark.createDataFrame(stuRDD,Student.class);  
         //stuDf.select("id","name","age").write().mode(SaveMode.Append).parquet("par");  //对文件指定列名  
         stuDf.printSchema();  
         stuDf.createOrReplaceTempView("student");  
         Dataset<Row> nameDf=spark.sql("select name from student where age<20");  
         nameDf.show();  
         //nameDf.write().mode("Append").text("name");//将查询结果写入一个文件  
     }  
     /** 
      * 动态转换 
      * @param spark 
      */  
     public static void dynamicCreate(SparkSession spark)  
     {  
         JavaRDD<Row> rows=spark.read().textFile("student.txt").javaRDD().map(new Function<String, Row>() {  
   
             public Row call(String line) throws Exception {  
                   
                 String[] parts=line.split(",");  
                 return RowFactory.create(  
                         Integer.valueOf(parts[0]),  
                         parts[1],  
                         Integer.valueOf(parts[2]));  
             }  
         });  
           
         //动态创建元数据  
         List<StructField> fields=new ArrayList<StructField>();  
         fields.add(DataTypes.createStructField("id", DataTypes.IntegerType, true));  
         fields.add(DataTypes.createStructField("name", DataTypes.StringType, true));  
         fields.add(DataTypes.createStructField("age", DataTypes.IntegerType, true));  
         StructType schema=DataTypes.createStructType(fields);  
         Dataset<Row> stuDf=spark.createDataFrame(rows, schema);  
         //stuDf.select("id","name","age").write().mode(SaveMode.Append).parquet("par");  
         stuDf.printSchema();  
         stuDf.createOrReplaceTempView("student");  
         Dataset<Row> nameDf=spark.sql("select name from student where age<20");  
         nameDf.show();  
         //nameDf.write().mode("Append").text("age");//将查询结果写入一个文件  
     }  
 }  

scala版本：

[plain]view plaincopy 
   
 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以前的版本。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/weixin_40725706/article/detail/589135