大数据Spark RDD运行设计底层原理

package main.scala.com.web.zhangyong168.cn.spark.test
import java.util.Properties
import com.web.zhangyong168.cn.spark.util.{PropertiesUtils, SparkTool}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Column, DataFrame, Row}
import org.slf4j.{Logger, LoggerFactory}
/** *
 *
 * @description Spark中DataFrame转RDD在转新的DataFrame
 * @author 张勇
 * @version 0.0.1
 * @date 2020年08月15日下午20:23:39
 */
object SparkRddMethod {
  val log: Logger = LoggerFactory.getLogger(getClass)
  val spark = SparkTool.getSparkSession()
   spark.sql("show databases").show()
  val pro: Properties = PropertiesUtils.loadProps("jdbc.properties")
  val map = Map[String, String](
    "driver" -> pro.getProperty("mysql.driver"),
    "url" -> pro.getProperty("mysql.url"),
    "user" -> pro.getProperty("mysql.user"),
    "password" -> pro.getProperty("mysql.password"),
    "dbtable" -> "tb_address_district_copy"
  )
  val df = spark.read.format("jdbc").options(map).load


  def main(args: Array[String]): Unit = {
    df.show()
    dataDf(df)
  }




  def  dataDf(df:DataFrame): Unit ={
    log.info("========================打印将原来的df所有的类型转成String类型begin========================")
    val cols: Array[String] = df.columns
    val colsd: Array[Column] = cols.map(f => df(f).cast(StringType))
    val df2: DataFrame = df.select(colsd: _*)
    df2.printSchema()
    log.info("========================打印将原来的df所有的类型转成String类型end========================")
    val dataRdd:RDD[Row]=df2.rdd
        .filter(r => r.getString(2)=="285")
      .groupBy(r => r.getString(2))
      .flatMap(obj =>{
        val row:Iterable[Row]=obj._2
        var list:List[Row]=List()
        var r1:Row=null
        row.foreach(r =>{ //注意类型要跟遍历的df类型一致
//          val id=r.getLong(0)
//          log.info("打印数据id："+id)
//          val district_name=r.getString(1)
//          log.info("打印数据district_name："+district_name)
//          val city_id=r.getLong(2)
//          log.info("打印数据city_id："+city_id)
//          val isdel=r.getInt(3)
//          log.info("打印数据isdel："+isdel)
//          val intime=r.getTimestamp(4)
//          log.info("打印数据intime："+intime)
          //test(id)
          r1=r
          list=list.+:(r1)
          log.info("打印数据r："+r)
        })
        log.info("打印数据list："+list)
        list.distinct
      })


    log.info("打印类型："+df.schema)
    df.printSchema()


    val schemaString="id,district_name,city_id,isdel,intime"
    val newDF=spark.createDataFrame(dataRdd,getSchema(schemaString))
    log.info("========================================================打印新集合的数据========================================================================")
    log.info("打印新数据总数："+ newDF.count())


    newDF.filter("city_id='285' and id='10'").sort("id").show()
  }


  /**
   * 更改要创建的df里面每个字段的不同类型
   */
  val schema:StructType = StructType(List(
    StructField("id", LongType, true),
    StructField("district_name", StringType, true),
    StructField("city_id", LongType, true),
    StructField("isdel", IntegerType, true),
    StructField("intime", TimestampType, true)
  ))


  def test(id :Long): Unit ={
    println("打印id:"+id)
  }


  // 创建 StructType 来定义结构
   def getSchema(schemaString:String):StructType={
    val fields:Array[StructField] =schemaString.split(",").map(fileName => StructField(fileName,StringType,nullable = true))
    //转化为schema
    StructType(fields)
  }