Scala创建DataFrame的几种方式

1 通过数据源创建DF

原始数据:

{"name":"Tom","age":18},
{"name":"Alice","age":17}

步骤:

//读取目录文件
scala> val df = spark.read.json("file:///opt/module/spark/mycode/a.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

//创建临时视图
scala> df.createOrReplaceTempView("people")

//查看DF
scala> spark.sql("select * from people").show

//打印结果
+---+-----+
|age| name|
+---+-----+
| 18|  Tom|
| 17|Alice|
+---+-----+

2 通过反射机制创建DF

原始数据:

Tom,18
Alice,17

步骤:

//导入包,支持把一个RDD隐式转换为一个DataFrame
scala> import spark.implicits._ 

//1.创建样例类
case class People(name:String, age: Int)

//2.创建RDD
scala> val rdd = sc.textFile("file:///opt/module/spark/mycode/b.txt")
rdd: org.apache.spark.rdd.RDD[String] = file:///opt/module/spark/mycode/b.txt MapPartitionsRDD[1] at textFile at <console>:24

//3.使用map算子将每一行都创建一个People对象
scala> val df = rdd.map(_.split(",")).map(data=>People(data(0),data(1).toInt)).toDF
df: org.apache.spark.sql.DataFrame = [name: string, age: int]

//4.打印df
scala> df.show
+-----+---+
| name|age|
+-----+---+
|  Tom| 18|
|Alice| 17|
+-----+---+

注意,import spark.implicits._ 中的spark指的是SparkSession对象(在scala交互式环境中默认存在了一个SparkSession对象)。

启动spark-shell时的信息:
在这里插入图片描述

3 使用编程定义方式创建DF

当无法提前定义case class时,就需要采用编程方式定义RDD模式

//导入相关包
scala> import org.apache.spark.sql.types._
import org.apache.spark.sql.types._ 
scala> import org.apache.spark.sql.Row
import org.apache.spark.sql.Row 

//1.定义一个模式字符串
scala> val schemaString = "name age"
schemaString: String = name age

//2.根据模式字符串创建模式
scala> val fields = schemaString.split(" ").map(fieldName=>StructField(fieldName,StringType,nullable=true))
fields: Array[org.apache.spark.sql.types.StructField] = Array(StructField(name,StringType,true), StructField(age,StringType,true))

scala> val schema = StructType(fields)
schema: org.apache.spark.sql.types.StructType = StructType(StructField(name,StringType,true), StructField(age,StringType,true))

//3.将rdd的每条数据转换成Row类型
scala> val rowRDD = rdd.map(_.split(",")).map(data=>Row(data(0),data(1)))
rowRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[19] at map at <console>:29

//4.根据设计的模式创建DF
scala> val df = spark.createDataFrame(rowRDD, schema)
df: org.apache.spark.sql.DataFrame = [name: string, age: string]

//5.打印
scala> df.show
+-----+---+                                                                     
| name|age|
+-----+---+
|  Tom| 18|
|Alice| 17|
+-----+---+
  • 1
    点赞
  • 6
    收藏
    觉得还不错? 一键收藏
  • 打赏
    打赏
  • 0
    评论

“相关推荐”对你有帮助么?

  • 非常没帮助
  • 没帮助
  • 一般
  • 有帮助
  • 非常有帮助
提交
评论
添加红包

请填写红包祝福语或标题

红包个数最小为10个

红包金额最低5元

当前余额3.43前往充值 >
需支付:10.00
成就一亿技术人!
领取后你会自动成为博主和红包主的粉丝 规则
hope_wisdom
发出的红包

打赏作者

每天进步一点丶

你的鼓励将是我创作的最大动力

¥1 ¥2 ¥4 ¥6 ¥10 ¥20
扫码支付:¥1
获取中
扫码支付

您的余额不足,请更换扫码支付或充值

打赏作者

实付
使用余额支付
点击重新获取
扫码支付
钱包余额 0

抵扣说明:

1.余额是钱包充值的虚拟货币,按照1:1的比例进行支付金额的抵扣。
2.余额无法直接购买下载,可以购买VIP、付费专栏及课程。

余额充值