我只是想知道在Apache Spark中RDD和DataFrame (Spark 2.0.0 DataFrame只是数据集[行]的类型别名)之间的区别是什么?
你能把一个转换成另一个吗?
我只是想知道在Apache Spark中RDD和DataFrame (Spark 2.0.0 DataFrame只是数据集[行]的类型别名)之间的区别是什么?
你能把一个转换成另一个吗?
当前回答
所有(RDD、DataFrame和DataSet)在一张图片中。
图片致谢
RDD
RDD是可以并行操作的元素的容错集合。
DataFrame
DataFrame是一个被组织成命名列的数据集。它是 概念上等价于关系数据库中的表或数据 框架,但是在底层有更丰富的优化。
数据集
数据集是数据的分布式集合。Dataset是Spark 1.6中新增的接口,提供rdd的优点 (强类型,能够使用强大的lambda函数) Spark SQL优化执行引擎的好处。 注意: 在Scala/Java中,Dataset of Rows (Dataset[Row])通常被称为DataFrames。
用一个代码片段对它们进行了很好的比较。
源
问:你能把一个转换成另一个,像RDD到DataFrame,反之亦然?
是的,两者都有可能
1. 使用.toDF() RDD到DataFrame
val rowsRdd: RDD[Row] = sc.parallelize(
Seq(
Row("first", 2.0, 7.0),
Row("second", 3.5, 2.5),
Row("third", 7.0, 5.9)
)
)
val df = spark.createDataFrame(rowsRdd).toDF("id", "val1", "val2")
df.show()
+------+----+----+
| id|val1|val2|
+------+----+----+
| first| 2.0| 7.0|
|second| 3.5| 2.5|
| third| 7.0| 5.9|
+------+----+----+
在Spark中将RDD对象转换为Dataframe
2. 使用.rdd()方法将DataFrame/DataSet转换为RDD
val rowsRdd: RDD[Row] = df.rdd() // DataFrame to RDD
其他回答
因为DataFrame是弱类型的,开发人员没有得到类型系统的好处。例如,假设你想从SQL中读取一些东西,并对其运行一些聚合:
val people = sqlContext.read.parquet("...")
val department = sqlContext.read.parquet("...")
people.filter("age > 30")
.join(department, people("deptId") === department("id"))
.groupBy(department("name"), "gender")
.agg(avg(people("salary")), max(people("age")))
当你说people("deptId")时,你得到的不是Int或Long对象,你得到的是你需要操作的Column对象。在具有丰富类型系统的语言(如Scala)中,您最终失去了所有类型安全,这增加了在编译时可以发现的运行时错误的数量。
相反,输入数据集[T]。当你这样做时:
val people: People = val people = sqlContext.read.parquet("...").as[People]
您实际上得到了一个People对象,其中deptId是一个实际的整型而不是列型,从而利用了类型系统。
从Spark 2.0开始,DataFrame和DataSet api将是统一的,其中DataFrame将是DataSet[Row]的类型别名。
所有(RDD、DataFrame和DataSet)在一张图片中。
图片致谢
RDD
RDD是可以并行操作的元素的容错集合。
DataFrame
DataFrame是一个被组织成命名列的数据集。它是 概念上等价于关系数据库中的表或数据 框架,但是在底层有更丰富的优化。
数据集
数据集是数据的分布式集合。Dataset是Spark 1.6中新增的接口,提供rdd的优点 (强类型,能够使用强大的lambda函数) Spark SQL优化执行引擎的好处。 注意: 在Scala/Java中,Dataset of Rows (Dataset[Row])通常被称为DataFrames。
用一个代码片段对它们进行了很好的比较。
源
问:你能把一个转换成另一个,像RDD到DataFrame,反之亦然?
是的,两者都有可能
1. 使用.toDF() RDD到DataFrame
val rowsRdd: RDD[Row] = sc.parallelize(
Seq(
Row("first", 2.0, 7.0),
Row("second", 3.5, 2.5),
Row("third", 7.0, 5.9)
)
)
val df = spark.createDataFrame(rowsRdd).toDF("id", "val1", "val2")
df.show()
+------+----+----+
| id|val1|val2|
+------+----+----+
| first| 2.0| 7.0|
|second| 3.5| 2.5|
| third| 7.0| 5.9|
+------+----+----+
在Spark中将RDD对象转换为Dataframe
2. 使用.rdd()方法将DataFrame/DataSet转换为RDD
val rowsRdd: RDD[Row] = df.rdd() // DataFrame to RDD
通过谷歌搜索“DataFrame definition”可以很好地定义一个DataFrame:
数据帧是一种表格,或者是一种二维的类似数组的结构 每一列包含对一个变量的测量,以及每一行 包含一个大小写。
因此,由于其表格格式,DataFrame具有额外的元数据,这允许Spark在最终查询上运行某些优化。
另一方面,RDD只是一个弹性分布式数据集(Resilient Distributed Dataset),它更像是一个数据黑箱,不能对其进行优化,因为可以对其执行的操作不受约束。
然而,你可以通过RDD方法从一个DataFrame到一个RDD,你也可以通过toDF方法从一个RDD到一个DataFrame(如果RDD是一个表格格式)
一般来说,由于内置的查询优化,建议尽可能使用DataFrame。
DataFrame相当于RDBMS中的表,也可以以类似于rdd中的“原生”分布式集合的方式进行操作。与rdd不同,dataframe跟踪模式并支持各种关系操作,从而实现更优化的执行。 每个DataFrame对象表示一个逻辑计划,但由于它们的“惰性”性质,直到用户调用特定的“输出操作”才会执行。
简单地说,RDD是核心组件,而DataFrame是spark 1.30引入的API。
RDD
数据分区的集合,称为RDD。这些RDD必须遵循以下几个属性:
不可变的, 容错, 分布式的, 更多。
这里RDD是结构化的或非结构化的。
DataFrame
DataFrame是Scala、Java、Python和r中可用的API,它允许处理任何类型的结构化和半结构化数据。要定义DataFrame,一个被组织成命名列的分布式数据集合,称为DataFrame。您可以很容易地优化DataFrame中的rdd。 您可以使用DataFrame一次处理JSON数据,parquet数据,HiveQL数据。
val sampleRDD = sqlContext.jsonFile("hdfs://localhost:9000/jsondata.json")
val sample_DF = sampleRDD.toDF()
这里Sample_DF被认为是DataFrame。sampleRDD(原始数据)称为RDD。