今天就跟大家聊聊有关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
lineRDD =sc.textFile("C:\\Users\\Administrator\\Desktop\\studentData.txt"); JavaRDD
rowJavaRDD = lineRDD.map(new Function
() { @Override
public Row call(String line) throws Exception {
String[] splitLine = line.split(",");
return RowFactory.create(Integer.valueOf(splitLine[0])
,Integer.valueOf(splitLine[1])
,splitLine[2]);
}
});
List
structFields=new ArrayList (); /*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
dataFrame = sqlContext.createDataFrame(rowJavaRDD, structType);
dataFrame.registerTempTable("studentInfo");
Dataset
resultDataSet = sqlContext.sql("select * from studentInfo where age > 17");
List
collect = resultDataSet.javaRDD().collect();
for(Row row: collect){
System.out.println(row);
}
sc.close();
}
}
pom.xml关键依赖:
2.2.0
2.11.8
org.apache.spark spark-sql_2.11 ${spark.version} org.apache.spark spark-core_2.11 ${spark.version}
看完上述内容,你们对Spark2.2.0中RDD转DataFrame的方式是什么有进一步的了解吗?如果还想了解更多知识或者相关内容,请关注创新互联行业资讯频道,感谢大家的支持。