scala spark 创建DataFrame的五种方式

灰太狼 2022-10-24 01:35 708阅读 0赞

**scala spark 创建DataFrame的多种方式**

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

import org.apache.log4j.{Level, Logger}
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
    import org.apache.spark.sql.{DataFrame, Row, SparkSession}
    
    /**
     *通过RDD[Row]和StructType创建DataFrame
     */
    object DataFrameDemo {
      def main(args: Array[String]): Unit = {
        Logger.getLogger("org").setLevel(Level.OFF)
        // 通过SparkSession创建spark入口
        val spark: SparkSession = SparkSession.builder()
          .appName(this.getClass.getSimpleName)
          .master("local[*]")
          .getOrCreate()
        println("1. 通过RDD[Row]和StructType创建")
        //1. 通过RDD[Row]和StructType创建
        val sparkRdd1: RDD[String] = spark.sparkContext.parallelize(List("X,22,M", "y,21,W", "N,22,M"))
        // 将RDD与Row联合
        val rowRdd: RDD[Row] = sparkRdd1.map(t => {
          val per: Array[String] = t.split(",")
          Row(per(0), per(1).toInt, per(2))
        })
        // 创建StructType实例,设置字段名和类型
        val schema: StructType = StructType(
          List(
            StructField("name", StringType),
            StructField("age", IntegerType),
            StructField("sex", StringType)
          )
        )
        // 创建dataFrame
        val dataFrame: DataFrame = spark.createDataFrame(rowRdd, schema)
        // 展示数据
        dataFrame.show()
        // 释放资源
        spark.stop()
      }
    }

输出结果

1. 通过RDD[Row]和StructType创建
    +----+---+---+
    |name|age|sex|
    +----+---+---+
    |   X| 22|  M|
    |   y| 21|  W|
    |   N| 22|  M|
    +----+---+---+

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

import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.{DataFrame, SparkSession}
    
    /**
     * 通过RDD.toDF直接创建DataFrame
     */
    object DataFrameDemo2 {
      def main(args: Array[String]): Unit = {
        Logger.getLogger("org").setLevel(Level.OFF)
        // 通过SparkSession创建spark入口
        val spark: SparkSession = SparkSession.builder()
          .appName(this.getClass.getSimpleName)
          .master("local[*]")
          .getOrCreate()
        println("2. 直接通过Rdd.toDF()创建DataFrame")
        // 创建RDD
        val sparkRdd3: RDD[String] = spark.sparkContext.parallelize(List("X,22,M", "y,21,W", "N,22,M"))
        // 将RDD与Row联合
        val toDFRdd: RDD[(String, Int, String)] = sparkRdd3.map(t => {
          val per: Array[String] = t.split(",")
          (per(0), per(1).toInt, per(2))
        })
        import org.apache.spark.sql.functions._
        import spark.implicits._
        // 创建DataFrame,设置schema字段名
        //写法1
      //val frame: DataFrame = toDFRdd.toDF("name", "age", "sex")
        //写法2
        val array =Array("name", "age", "sex")
        val frame: DataFrame = toDFRdd.toDF(array: _*)
    
        // 展示数据
        frame.agg(sum("age") as "avg_age").show()
        // 释放资源
        spark.stop()
      }
    
    }

输出结果

2. 直接通过Rdd.toDF()创建DataFrame
    +-------+
    |avg_age|
    +-------+
    |     65|
    +-------+

## 3. 通过RDD和ScalaBean创建DataFrame ##

import org.apache.log4j.{Level, Logger}
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.{DataFrame, SparkSession}
    
    /**
     * 通过RDD和scalabean创建DataFrame
     */
    object DataFrameDemo3 {
      def main(args: Array[String]): Unit = {
        Logger.getLogger("org").setLevel(Level.OFF)
        // 通过SparkSession创建spark入口
        val spark: SparkSession = SparkSession.builder()
          .appName(this.getClass.getSimpleName)
          .master("local[*]")
          .getOrCreate()
        println("3. 通过RDD和ScalaBean创建DataFrame")
        val sparkRdd: RDD[String] = spark.sparkContext.parallelize(List("X,22,M", "y,21,W", "N,22,M"))
        // 关联SacalaBean的RDD
        val beanRdd: RDD[Per] = sparkRdd.map(t => {
          val per: Array[String] = t.split(",")
          Per(per(0), per(1).toInt, per(2))
        })
        // 必须导入隐式转换才能使用.toDF
        import spark.implicits._
        // 创建dataFrame
        val df: DataFrame = beanRdd.toDF
        // 创建视图
        df.createTempView("t_per")
        // 查询数据
        val res: DataFrame = spark.sql("SELECT name,age FROM t_per ORDER BY age")
        // 展示数据
        res.show()
        // 释放资源
        spark.stop()
      }
    
      //ScalaBean
      case class Per(name: String, age: Int, sex: String)
    }

输出结果

3. 通过RDD和ScalaBean创建DataFrame
    +----+---+
    |name|age|
    +----+---+
    |   y| 21|
    |   N| 22|
    |   X| 22|
    +----+---+

## 4. 通过RDD联合JavaBean创建DataFrame ##

import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.{DataFrame, SparkSession}
    
    /**
     * 通过RDD和javabean创建DataFrame
     */
    object DataFrameDemo4 {
      def main(args: Array[String]): Unit = {
        Logger.getLogger("org").setLevel(Level.OFF)
        // 通过SparkSession创建spark入口
        val spark: SparkSession = SparkSession.builder()
          .appName(this.getClass.getSimpleName)
          .master("local[*]")
          .getOrCreate()
        println("4. 通过RDD联合JavaBean创建DataFrame")
        // 创建RDD
        val sparkRdd4: RDD[String] = spark.sparkContext.parallelize(List("X,22,M", "y,21,W", "N,22,M"))
        // 将RDD与Row联合
        val javaBeanRdd: RDD[Person] = sparkRdd4.map(t => {
          val per: Array[String] = t.split(",")
          new Person(per(0), per(1).toInt, per(2))
        })
        // 创建DataFrame
        val frame1: DataFrame = spark.createDataFrame(javaBeanRdd, classOf[Person])
        // 展示数据
        frame1.show()
        // 释放资源
        spark.stop()
      }
    }

Java bean class

public class Person {
        private String name;
        private int age;
        private String sex;
    
        public Person(String name, int age, String sex) {
            this.name = name;
            this.age = age;
            this.sex = sex;
        }
    
        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;
        }
    
        public String getSex() {
            return sex;
        }
    
        public void setSex(String sex) {
            this.sex = sex;
        }
    }

输出结果

4. 通过RDD联合JavaBean创建DataFrame
    +---+----+---+
    |age|name|sex|
    +---+----+---+
    | 22|   X|  M|
    | 21|   y|  W|
    | 22|   N|  M|
    +---+----+---+

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

import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.{DataFrame, SparkSession}
    
    /**
     * 通过RDD和javabean创建DataFrame
     */
    object DataFrameDemo5 {
      def main(args: Array[String]): Unit = {
        Logger.getLogger("org").setLevel(Level.OFF)
        // 通过SparkSession创建spark入口
        val spark: SparkSession = SparkSession.builder()
          .appName(this.getClass.getSimpleName)
          .master("local[*]")
          .getOrCreate()
        println("5、手写的示例5")
        val rdd11 = spark.sparkContext.parallelize(Seq(
          Row(8, "bat"),
          Row(64, "mouse"),
          Row(-27, "horse")
        ))
        rdd11.foreach(println)
       // 创建
        val newRDD: RDD[NumberWord]= rdd11.map(x=>{
          val rdto = convertNumberWordDTO(x.getInt(0)+100,x.getString(1))
          rdto
        })
        val ds = spark.createDataFrame(newRDD, classOf[NumberWord])
        ds.show(100,false)
    
        println(ds.schema) //StructType(StructField(number,IntegerType,false), StructField(word,StringType,true))
        //调用 someDF.schema.printTreeString()得到someDF的schema:
        ds.schema.printTreeString()
        //root
        // |-- number: integer (nullable = false)
        // |-- word: string (nullable = true)
    
        // 释放资源
        spark.stop()
      }
      def convertNumberWordDTO(number: Int,word:String) ={
        val rdto = new NumberWord()
        rdto.setNumber(number)
        rdto.setWord(word)
        rdto
      }
    }

Java bean class

public class NumberWord {
        private int number;
        private String word;
    
        public NumberWord() {
        }
    
        public NumberWord(int number, String word) {
            this.number = number;
            this.word = word;
        }
    
        public int getNumber() {
            return number;
        }
    
        public void setNumber(int number) {
            this.number = number;
        }
    
        public String getWord() {
            return word;
        }
    
        public void setWord(String word) {
            this.word = word;
        }
    }

输出结果