第60课：使用Java实战RDD与DataFrame动态转换操作

在企业中目前应用比较多的就是动态转换。动态的意思：是指提前不知道RDD中的每个record的列的个数，以及列的内容，只有在运行的时候才会知道，Row代表table中的一行数据。

实战代码如下：

package com.dt.spark.SparkApps.sql;
/**
* 打印结果：
[1,Spark,7]
[2,Hadoop,10]
[3,Flink,5]
*/
import java.util.ArrayList;
import java.util.List;

import org.apache.spark.SparkConf;
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.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

public class RDDToDataFrameByProgrammatically {

public static void main(String[] args) {
SparkConf conf = new SparkConf().setMaster("local").
setAppName("RDDToDataFrameByReflection");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);

//读取数据
JavaRDD<String> lines = sc.textFile("E://persons.txt");

/**
* 第一步：在RDD的基础上创建类型为Row的RDD
*/
//首先，必须将RDD变成以Row为类型的RDD。Row可以简单理解为Table的一行数据
JavaRDD<Row> personsRDD = lines.map(new Function<String,Row>(){

@Override
public Row call(String line) throws Exception {
String[] splited = line.split(",");
return RowFactory.create(Integer.valueOf(splited[0]),splited[1],Integer.valueOf(splited[2]));
}

});

/**
* 第二步：动态构造DataFrame的元数据，一般而言，有多少列以及每列的具体类型可能来自于
* JSON文件，也可能来自于DB
*/
//对Row具体指定元数据信息。
List<StructField> structFields = new ArrayList<StructField>();
//列名称  列的具体类型（Integer Or String） 是否为空一般为true，实际在开发环境是通过for循环，而不是手动添加
structFields.add(DataTypes.createStructField("id", DataTypes.IntegerType, true));
structFields.add(DataTypes.createStructField("name", DataTypes.StringType, true));
structFields.add(DataTypes.createStructField("age", DataTypes.IntegerType, true));
//构建StructType,用于最后DataFrame元数据的描述
StructType structType = DataTypes.createStructType(structFields);

/**
* 第三步：基于已有的MetaData以及RDD<Row>来构造DataFrame
*/
DataFrame personsDF = sqlContext.createDataFrame(personsRDD, structType);

/**
* 第四步：注册成临时表以供后续的SQL查询操作
*/
personsDF.registerTempTable("persons");

/**
* 第五步：进行数据的多维度分析
*/
DataFrame result = sqlContext.sql("select * from persons");
/**
* 第六步：对结果进行处理，包括由DataFrame转换成为RDD<Row>,以及结果的持久化
*/
List<Row> listRow = result.javaRDD().collect();
for(Row row : listRow){
System.out.println(row);
}

}

}

本课程笔记来源于：