DataFrame 和 RDD互操作的两种方式

1）反射 ：case class 前提 事先需要知道你的字段、字段类型
2）编程 : Row 事先不知道列
优先选择反射 ，下面放两个实例

package bl.test.spark

import org.apache.spark.sql.types.{StructField, StructType,StringType,IntegerType}
import org.apache.spark.sql.{Row, SparkSession}

object DataFrameRDDApp   {
  def main(args: Array[String]) {

    val spark = SparkSession.builder().appName("DataFrameRDDApp")
      .master("local[2]").getOrCreate()
    inferReflection(spark)
    program(spark)
    spark.stop()
  }

  def program(spark: SparkSession): Unit ={
    val rdd = spark.sparkContext.textFile("file:////home/zy/Desktop/success.txt")
    val infoRDD = rdd.map(_.split(",")).map(line => Row(line(0).toInt, line(1), line(2).toInt))
    val structType = StructType(Array(
      StructField("id",IntegerType,true),
      StructField("name",StringType,true),
      StructField("age",IntegerType,true)

    ));
    val infoDF = spark.createDataFrame(infoRDD,structType);
    infoDF.printSchema()
    infoDF.show()
    infoDF.filter((infoDF.col(("age")) > 30)).show()
    //创建一张表 使用基于sql的api
    infoDF.createOrReplaceTempView("infos")
    spark.sql("select * from infos where age > 30").show()
  }


  def inferReflection(spark: SparkSession) = {
    //RDD ==> DataFrame
    val rdd = spark.sparkContext.textFile("file:////home/zy/Desktop/success.txt")
    //导入隐式转换
    import spark.implicits._
    //按照逗号切割
    val infoDF = rdd.map(_.split(",")).map(line => Info(line(0).toInt, line(1), line(2).toInt)).toDF()
    infoDF.show()
    infoDF.filter((infoDF.col(("age")) > 30)).show()
    //创建一张表 使用基于sql的api
    infoDF.createOrReplaceTempView("infos")
    spark.sql("select * from infos where age > 30").show()
  }

  case class Info(id:Int, name:String, age:Int)
}