一、創(chuàng)建DataSet

DataSet與RDD相似，但是，它們不使用Java序列化或Kryo，而是使用專用的Encoder對對象進(jìn)行序列化以進(jìn)行網(wǎng)絡(luò)處理或傳輸。雖然編碼器和標(biāo)準(zhǔn)序列化都負(fù)責(zé)將對象轉(zhuǎn)換為字節(jié)，但是編碼器是動(dòng)態(tài)生成的代碼，并使用一種格式，該格式允許Spark執(zhí)行許多操作，例如過濾，排序和哈希處理，而無需將字節(jié)反序列化回對象。

Encoder<Man> manEncoder = Encoders.bean(Man.class); Dataset<Man> manDS = spark.createDataset(Collections.singletonList(man),manEncoder );Encoder<String> strEncoder = Encoders.STRING(); Dataset<String> strDS = spark.createDataset(Arrays.asList("a", "b", "c"), strEncoder);

?

二、RDD轉(zhuǎn)DataSet

Spark SQL支持兩種將現(xiàn)有RDD轉(zhuǎn)換為數(shù)據(jù)集的方法。第一種方法使用反射來推斷包含特定對象類型的RDD的架構(gòu)。這種基于反射的方法可以使代碼更簡潔，并且在編寫Spark應(yīng)用程序時(shí)已經(jīng)了解架構(gòu)時(shí)可以很好地工作。

創(chuàng)建數(shù)據(jù)集的第二種方法是通過編程界面，該界面允許您構(gòu)造模式，然后將其應(yīng)用于現(xiàn)有的RDD。盡管此方法較為冗長，但可以在運(yùn)行時(shí)才知道列及其類型的情況下構(gòu)造數(shù)據(jù)集。

（1）使用反射推斷表的模式

Spark SQL支持自動(dòng)將JavaBean的RDD 轉(zhuǎn)換為DataFrame，使用反射得到Bean信息，定義了表的模式。Spark SQL不支持包含Map字段的JavaBean 。 不過，支持嵌套JavaBean和List或Array字段。

可以通過創(chuàng)建一個(gè)實(shí)現(xiàn)Serializable并為其所有字段具有g(shù)etter和setter的類來創(chuàng)建JavaBean。

JavaRDD<Man> manRDD = spark.read().textFile("/test.txt").javaRDD().map(line -> {String[] splits = line.split(",");Man man = new Man();man.setName(splits[0]);man.setDesc(splits[1]);return man;});Dataset<Row> manDF = spark.createDataFrame(manRDD, Man.class);

?（2）以編程方式指定架構(gòu)

當(dāng)無法提前定義JavaBean類時(shí)（例如，記錄的結(jié)構(gòu)編碼為字符串，或者將解析文本數(shù)據(jù)集，并且為不同的用戶設(shè)計(jì)不同的字段），Dataset<Row>可以通過三個(gè)步驟以編程方式創(chuàng)建a?。

Row從原始RDD創(chuàng)建一個(gè)的RDD；

在第1步中創(chuàng)建的RDD中創(chuàng)建StructType與Rows的結(jié)構(gòu)匹配?表示的模式。

Row通過通過createDataFrame提供的方法將架構(gòu)應(yīng)用于的RDD?SparkSession。

JavaRDD<String> manRDD = ... String cols = "name,desc"; List<StructField> fields = new LinkedList<>(); for (String fieldName : cols.split(",")) {StructField field = DataTypes.createStructField(fieldName, DataTypes.StringType, true);fields.add(field); } StructType schema = DataTypes.createStructType(fields); JavaRDD<Row> rowRDD = manRDD.map((Function<String, Row>) record -> {String[] splits = record.split(",");return RowFactory.create(splits[0], splits[1].trim()); });Dataset<Row> manDataFrame = spark.createDataFrame(rowRDD, schema);

?

三、DataSet轉(zhuǎn)RDD

?toJavaRDD方法直接將DataSet轉(zhuǎn)RDD

JavaRDD<Row> javaRDD = df.toJavaRDD();

?

總結(jié)

以上是生活随笔為你收集整理的Spark SQL（四）之DataSet与RDD转换的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。