spark将RDD转换为DataFrame

1. 方法一（不推荐）

首先说明下面所有方法我都是使用scala语言实现的，python实际上类似，编码方式基本一致的。spark将csv转换为DataFrame,可以先文件读取为RDD，然后再进行map操作，对每一行进行分割。
再将schema和rdd分割后的Rows回填，sparkSession创建的dataFrame

 1 val spark = SparkSession
 2      .builder()
 3      .appName("sparkdf")
 4      .master("local[1]")
 5      .getOrCreate()
 6
 7       //设置spark的上下文sparkContext
 8      val sc = spark.sparkContext
 9      val fileRDD = sc.textFile("/home/hadoop/Downloads/filesmall2.csv")
10      //val rdd = fileRDD.filter(line => line.split("\t").length != 30)
11      val df = spark.createDataFrame(fileRDD.map(line=>HttpSchema.parseLog(line)),HttpSchema.struct)
12      df.show(3)

这里的RDD是通过读取文件创建的所以也可以看做是将RDD转换为DataFrame

 1object HttpSchema {
 2
 3  def parseLog(x:String): Row = {
 4
 5    var fields = x.split("\t")
 6
 7    val _id = fields(0)
 8    val srcIp = fields(1)
 9    val srcPort = fields(2)
10    //这种方法比较麻烦的地方是row里面的字段名要和struct中的字段对应上
11    RowFactory.create(_id,srcIp,srcPort)
12  }
13
14  //设置schema描述
15  val struct = StructType(
16    Array( StructField("_id",StringType),
17      StructField("srcIp",StringType),
18      StructField("srcPort",StringType),
19    )
20  )
21}

这也是这种方法不推荐使用的地方，因为返回的Row中的字段名要与schema中的字段名要一致，当字段多于22个这个需要集成一个

2. 方法二
   //使用隐式转换的方式来进行转换

 1val spark = SparkSession
 2      .builder()
 3      .appName("sparkdf")
 4      .master("local[1]")
 5      .getOrCreate()
 6
 7      //使用隐式转换必须导入这个才可以使用只有import spark.implicits._之后，RDD才有toDF、toDS功能
 8
 9
10      import spark.implicits._
11
12       //设置spark的上下文sparkContext
13      val sc = spark.sparkContext
14
15      val fileRDD = sc.textFile("/home/hadoop/Downloads/filesmall2.csv")
16
17      case class HttpClass(id:String,srcIp:String,srcPort:Int)
18
19      val df = fileRDD.map(_.split("\t")).map(line=>HttpClass(line(0),line(1),line(2).toInt)).toDF()

当然也可以不创建类对象

1rdd.map{x=>val par=x.split(",");(par(0),par(1).toInt)}.toDF("name","age")

dataFrame转换为RDD只需要将collect就好，df.collect   RDD[row]类型，就可以按row取出

spark读取csv转化为DataFrame

1. 方法一

 1 val conf = new SparkConf().setAppName("word count").setMaster("local[1]")
 2
 3    val sc = new SparkContext(conf)
 4    println("spark version: " + sc.version)
 5
 6    val spark = new SQLContext(sc)
 7
 8    import spark.implicits._
 9
10    val df = spark.read.format("com.databricks.spark.csv")
11      .option("header", "false")
12      .option("inferSchema", "false") //是否自动推到内容的类型
13      .option("delimiter",",")  //分隔符，默认为 ,
14      .load("/home/hadoop/Downloads/Salary_Data.csv")
15    df.show()
16
17
18    //进行写数据
19    data.repartition(1).write.format("com.databricks.spark.csv")
20      .option("header", "false")//在csv第一行有属性"true"，没有就是"false"
21      .option("delimiter",",")//默认以","分割
22      .save(outpath)
23     sparkContext.stop()

sparkContext.sql()操作完成后直接返回的是DataFrame
当然可以间接采用将csv直接转换为RDD然后再将RDD转换为DataFrame

2. 方法二

1// 读取数据并分割每个样本点的属性值 形成一个Array[String]类型的RDD 
2val rdd = sc.textFile("file:///home/xuqm/ML_Data/input/synthetic_control.data").map(_.split("\\s+")) 
3// 将rdd转换成LabeledPoint类型的RDD 
4val LabeledPointRdd = rdd.map(x=>LabeledPoint(0,Vectors.dense(x.map(_.toDouble))))
5// 转成DataFrame并只取"features"列 
6val data = spark.createDataFrame(LabeledPointRdd).select("features")