溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
今天就跟大家聊聊有關Spark2.2.0中RDD轉DataFrame的方式是什么，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內容，希望大家根據這篇文章可以有所收獲。
Spark SQL如何將現有的RDDs轉換為數據集。
方法：通過編程接口，該接口允許您構造一個模式，然后將其應用于現有的RDD。雖然此方法更詳細，但它允許您在列及其類型直到運行時才知道時構造數據集。
數據準備studentData.txt
1001,20,zhangsan1002,17,lisi1003,24,wangwu1004,16,zhaogang
代碼實例：
package com.unicom.ljs.spark220.study;import org.apache.spark.SparkConf;import org.apache.spark.SparkContext;import org.apache.spark.api.java.JavaRDD;import org.apache.spark.api.java.JavaSparkContext;import org.apache.spark.api.java.function.Function;import org.apache.spark.rdd.RDD;import org.apache.spark.sql.*;import org.apache.spark.sql.types.*;import java.util.ArrayList;import java.util.List;/*** @author: Created By lujisen* @company ChinaUnicom Software JiNan* @date: 2020-01-21 13:42* @version: v1.0* @description: com.unicom.ljs.spark220.study*/public class RDD2DataFrameProgramatically {public static void main(String[] args) {SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD2DataFrameProgramatically");JavaSparkContext sc = new JavaSparkContext(sparkConf);SQLContext sqlContext = new SQLContext(sc);JavaRDD<String> lineRDD =sc.textFile("C:\\Users\\Administrator\\Desktop\\studentData.txt");JavaRDD<Row> rowJavaRDD = lineRDD.map(new Function<String, Row>() {@Overridepublic Row call(String line) throws Exception {String[] splitLine = line.split(",");return RowFactory.create(Integer.valueOf(splitLine[0]),Integer.valueOf(splitLine[1]),splitLine[2]);}});List<StructField> structFields=new ArrayList<StructField>();/*StructField structField1=new StructField("id", DataTypes.IntegerType,true);*/structFields.add(DataTypes.createStructField("id",DataTypes.IntegerType,true));structFields.add(DataTypes.createStructField("age",DataTypes.IntegerType,true));structFields.add(DataTypes.createStructField("name",DataTypes.StringType,true));StructType structType=DataTypes.createStructType(structFields);Dataset<Row> dataFrame = sqlContext.createDataFrame(rowJavaRDD, structType);dataFrame.registerTempTable("studentInfo");Dataset<Row> resultDataSet = sqlContext.sql("select * from studentInfo where age > 17");List<Row> collect = resultDataSet.javaRDD().collect();for(Row row: collect){System.out.println(row);}sc.close();}}
pom.xml關鍵依賴：
<spark.version>2.2.0</spark.version>
<scala.version>2.11.8</scala.version><dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>${spark.version}</version></dependency><dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.11</artifactId> <version>${spark.version}</version></dependency>
看完上述內容，你們對Spark2.2.0中RDD轉DataFrame的方式是什么有進一步的了解嗎？如果還想了解更多知識或者相關內容，請關注億速云行業資訊頻道，感謝大家的支持。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
