devbox
花生_TL007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

scala spark 创建DataFrame的五种方式_scala dataframe

作者：花生_TL007 | 2024-06-09 08:31:52

赞

踩

scala dataframe

scala spark 创建DataFrame的多种方式

1. 通过RDD[Row]和StructType创建

  1. import org.apache.log4j.{Level, Logger}
  2. import org.apache.spark.rdd.RDD
  3. import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
  4. import org.apache.spark.sql.{DataFrame, Row, SparkSession}
  5.  
  6. /**
  7.  *通过RDD[Row]和StructType创建DataFrame
  8.  */
  9. object DataFrameDemo {
  10.   def main(args: Array[String]): Unit = {
  11.     Logger.getLogger("org").setLevel(Level.OFF)
  12.     // 通过SparkSession创建spark入口
  13.     val spark: SparkSession = SparkSession.builder()
  14.       .appName(this.getClass.getSimpleName)
  15.       .master("local[*]")
  16.       .getOrCreate()
  17.     println("1. 通过RDD[Row]和StructType创建")
  18.     //1. 通过RDD[Row]和StructType创建
  19.     val sparkRdd1: RDD[String] = spark.sparkContext.parallelize(List("X,22,M", "y,21,W", "N,22,M"))
  20.     // 将RDD与Row联合
  21.     val rowRdd: RDD[Row] = sparkRdd1.map(t => {
  22.       val per: Array[String] = t.split(",")
  23.       Row(per(0), per(1).toInt, per(2))
  24.     })
  25.     // 创建StructType实例,设置字段名和类型
  26.     val schema: StructType = StructType(
  27.       List(
  28.         StructField("name", StringType),
  29.         StructField("age", IntegerType),
  30.         StructField("sex", StringType)
  31.       )
  32.     )
  33.     // 创建dataFrame
  34.     val dataFrame: DataFrame = spark.createDataFrame(rowRdd, schema)
  35.     // 展示数据
  36.     dataFrame.show()
  37.     // 释放资源
  38.     spark.stop()
  39.   }
  40. }

输出结果

  1. 1. 通过RDD[Row]和StructType创建
  2. +----+---+---+
  3. |name|age|sex|
  4. +----+---+---+
  5. |   X| 22|  M|
  6. |   y| 21|  W|
  7. |   N| 22|  M|
  8. +----+---+---+

2. 直接通过Rdd.toDF()创建DataFrame

  1. import org.apache.spark.rdd.RDD
  2. import org.apache.spark.sql.{DataFrame, SparkSession}
  3.  
  4. /**
  5.  * 通过RDD.toDF直接创建DataFrame
  6.  */
  7. object DataFrameDemo2 {
  8.   def main(args: Array[String]): Unit = {
  9.     Logger.getLogger("org").setLevel(Level.OFF)
  10.     // 通过SparkSession创建spark入口
  11.     val spark: SparkSession = SparkSession.builder()
  12.       .appName(this.getClass.getSimpleName)
  13.       .master("local[*]")
  14.       .getOrCreate()
  15.     println("2. 直接通过Rdd.toDF()创建DataFrame")
  16.     // 创建RDD
  17.     val sparkRdd3: RDD[String] = spark.sparkContext.parallelize(List("X,22,M", "y,21,W", "N,22,M"))
  18.     // 将RDD与Row联合
  19.     val toDFRdd: RDD[(String, Int, String)] = sparkRdd3.map(t => {
  20.       val per: Array[String] = t.split(",")
  21.       (per(0), per(1).toInt, per(2))
  22.     })
  23.     import org.apache.spark.sql.functions._
  24.     import spark.implicits._
  25.     // 创建DataFrame,设置schema字段名
  26.     //写法1
  27.   //val frame: DataFrame = toDFRdd.toDF("name", "age", "sex")
  28.     //写法2
  29.     val array =Array("name", "age", "sex")
  30.     val frame: DataFrame = toDFRdd.toDF(array: _*)
  31.  
  32.     // 展示数据
  33.     frame.agg(sum("age") as "avg_age").show()
  34.     // 释放资源
  35.     spark.stop()
  36.   }
  37.  
  38. }

输出结果

  1. 2. 直接通过Rdd.toDF()创建DataFrame
  2. +-------+
  3. |avg_age|
  4. +-------+
  5. |     65|
  6. +-------+

3. 通过RDD和ScalaBean创建DataFrame

  1. import org.apache.log4j.{Level, Logger}
  2. import org.apache.spark.rdd.RDD
  3. import org.apache.spark.sql.{DataFrame, SparkSession}
  4.  
  5. /**
  6.  * 通过RDD和scalabean创建DataFrame
  7.  */
  8. object DataFrameDemo3 {
  9.   def main(args: Array[String]): Unit = {
  10.     Logger.getLogger("org").setLevel(Level.OFF)
  11.     // 通过SparkSession创建spark入口
  12.     val spark: SparkSession = SparkSession.builder()
  13.       .appName(this.getClass.getSimpleName)
  14.       .master("local[*]")
  15.       .getOrCreate()
  16.     println("3. 通过RDD和ScalaBean创建DataFrame")
  17.     val sparkRdd: RDD[String] = spark.sparkContext.parallelize(List("X,22,M", "y,21,W", "N,22,M"))
  18.     // 关联SacalaBean的RDD
  19.     val beanRdd: RDD[Per] = sparkRdd.map(t => {
  20.       val per: Array[String] = t.split(",")
  21.       Per(per(0), per(1).toInt, per(2))
  22.     })
  23.     // 必须导入隐式转换才能使用.toDF
  24.     import spark.implicits._
  25.     // 创建dataFrame
  26.     val df: DataFrame = beanRdd.toDF
  27.     // 创建视图
  28.     df.createTempView("t_per")
  29.     // 查询数据
  30.     val res: DataFrame = spark.sql("SELECT name,age FROM t_per ORDER BY age")
  31.     // 展示数据
  32.     res.show()
  33.     // 释放资源
  34.     spark.stop()
  35.   }
  36.  
  37.   //ScalaBean
  38.   case class Per(name: String, age: Int, sex: String)
  39. }

输出结果

  1. 3. 通过RDD和ScalaBean创建DataFrame
  2. +----+---+
  3. |name|age|
  4. +----+---+
  5. |   y| 21|
  6. |   N| 22|
  7. |   X| 22|
  8. +----+---+

4. 通过RDD联合JavaBean创建DataFrame

  1. import org.apache.spark.rdd.RDD
  2. import org.apache.spark.sql.{DataFrame, SparkSession}
  3.  
  4. /**
  5.  * 通过RDD和javabean创建DataFrame
  6.  */
  7. object DataFrameDemo4 {
  8.   def main(args: Array[String]): Unit = {
  9.     Logger.getLogger("org").setLevel(Level.OFF)
  10.     // 通过SparkSession创建spark入口
  11.     val spark: SparkSession = SparkSession.builder()
  12.       .appName(this.getClass.getSimpleName)
  13.       .master("local[*]")
  14.       .getOrCreate()
  15.     println("4. 通过RDD联合JavaBean创建DataFrame")
  16.     // 创建RDD
  17.     val sparkRdd4: RDD[String] = spark.sparkContext.parallelize(List("X,22,M", "y,21,W", "N,22,M"))
  18.     // 将RDD与Row联合
  19.     val javaBeanRdd: RDD[Person] = sparkRdd4.map(t => {
  20.       val per: Array[String] = t.split(",")
  21.       new Person(per(0), per(1).toInt, per(2))
  22.     })
  23.     // 创建DataFrame
  24.     val frame1: DataFrame = spark.createDataFrame(javaBeanRdd, classOf[Person])
  25.     // 展示数据
  26.     frame1.show()
  27.     // 释放资源
  28.     spark.stop()
  29.   }
  30. }

Java bean class

  1. public class Person {
  2.     private String name;
  3.     private int age;
  4.     private String sex;
  5.  
  6.     public Person(String name, int age, String sex) {
  7.         this.name = name;
  8.         this.age = age;
  9.         this.sex = sex;
  10.     }
  11.  
  12.     public String getName() {
  13.         return name;
  14.     }
  15.  
  16.     public void setName(String name) {
  17.         this.name = name;
  18.     }
  19.  
  20.     public int getAge() {
  21.         return age;
  22.     }
  23.  
  24.     public void setAge(int age) {
  25.         this.age = age;
  26.     }
  27.  
  28.     public String getSex() {
  29.         return sex;
  30.     }
  31.  
  32.     public void setSex(String sex) {
  33.         this.sex = sex;
  34.     }
  35. }

输出结果

  1. 4. 通过RDD联合JavaBean创建DataFrame
  2. +---+----+---+
  3. |age|name|sex|
  4. +---+----+---+
  5. | 22|   X|  M|
  6. | 21|   y|  W|
  7. | 22|   N|  M|
  8. +---+----+---+

5. 手写的通过RDD和javabean创建DataFrame

  1. import org.apache.spark.rdd.RDD
  2. import org.apache.spark.sql.{DataFrame, SparkSession}
  3.  
  4. /**
  5.  * 通过RDD和javabean创建DataFrame
  6.  */
  7. object DataFrameDemo5 {
  8.   def main(args: Array[String]): Unit = {
  9.     Logger.getLogger("org").setLevel(Level.OFF)
  10.     // 通过SparkSession创建spark入口
  11.     val spark: SparkSession = SparkSession.builder()
  12.       .appName(this.getClass.getSimpleName)
  13.       .master("local[*]")
  14.       .getOrCreate()
  15.     println("5、手写的示例5")
  16.     val rdd11 = spark.sparkContext.parallelize(Seq(
  17.       Row(8, "bat"),
  18.       Row(64, "mouse"),
  19.       Row(-27, "horse")
  20.     ))
  21.     rdd11.foreach(println)
  22.    // 创建
  23.     val newRDD: RDD[NumberWord]= rdd11.map(x=>{
  24.       val rdto = convertNumberWordDTO(x.getInt(0)+100,x.getString(1))
  25.       rdto
  26.     })
  27.     val ds = spark.createDataFrame(newRDD, classOf[NumberWord])
  28.     ds.show(100,false)
  29.  
  30.     println(ds.schema) //StructType(StructField(number,IntegerType,false), StructField(word,StringType,true))
  31.     //调用 someDF.schema.printTreeString()得到someDF的schema:
  32.     ds.schema.printTreeString()
  33.     //root
  34.     // |-- number: integer (nullable = false)
  35.     // |-- word: string (nullable = true)
  36.  
  37.     // 释放资源
  38.     spark.stop()
  39.   }
  40.   def convertNumberWordDTO(number: Int,word:String) ={
  41.     val rdto = new NumberWord()
  42.     rdto.setNumber(number)
  43.     rdto.setWord(word)
  44.     rdto
  45.   }
  46. }

Java bean class

  1. public class NumberWord {
  2.     private int number;
  3.     private String word;
  4.  
  5.     public NumberWord() {
  6.     }
  7.  
  8.     public NumberWord(int number, String word) {
  9.         this.number = number;
  10.         this.word = word;
  11.     }
  12.  
  13.     public int getNumber() {
  14.         return number;
  15.     }
  16.  
  17.     public void setNumber(int number) {
  18.         this.number = number;
  19.     }
  20.  
  21.     public String getWord() {
  22.         return word;
  23.     }
  24.  
  25.     public void setWord(String word) {
  26.         this.word = word;
  27.     }
  28. }

输出结果

  1. 5、手写的示例5
  2. [8,bat]
  3. [64,mouse]
  4. [-27,horse]
  5. +------+-----+
  6. |number|word |
  7. +------+-----+
  8. |108   |bat  |
  9. |164   |mouse|
  10. |73    |horse|
  11. +------+-----+
  12.  
  13. StructType(StructField(number,IntegerType,false), StructField(word,StringType,true))
  14. root
  15.  |-- number: integer (nullable = false)
  16.  |-- word: string (nullable = true)

 

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

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号