使用反射的方式将RDD转换为DataFrame

Java：

package cn.spark.sql;

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.SQLContext;

import java.util.List;

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

        JavaRDD<String> lines = sc.textFile("C://Users//Desktop//students.txt");

        JavaRDD<Student> students = lines.map(
                new Function<String, Student>() {
                    private static final long serialVersionUID = 1L;
            @Override
            public Student call(String s) throws Exception {
                String[] lineSplit = s.split(",");
                Student stu = new Student();
                stu.setID(Integer.valueOf(lineSplit[0].trim()));
                stu.setName(lineSplit[1]);
                stu.setAge(Integer.valueOf(lineSplit[2].trim()));
                return stu;


            }
        });

        /*
        * 使用反射方式 将RDD转换为DataFrame
        * 将Student.class 传入进去 其实就是用反射的方式来创建DataFrame
        * 因为底层通过Student class 进行反射 来获取其中的field
        * 要求 javabean中需要实现serializable 接口
        * */
        DataFrame studentDF = sqlContext.createDataFrame(
                students,Student.class);
        //拿到一个DataFrame 后 就可以注册一个临时表
        studentDF.registerTempTable("students");
        DataFrame teengerDF = sqlContext.sql("select * from students where age <= 18");

        //将查询出来的DataFrame 再次转换成RDD
        JavaRDD<Row> teengerRDD = teengerDF.javaRDD();


        //将EDD中的数据 进行映射 映射为student
        JavaRDD<Student> teengerStudentRDD = teengerRDD.map(new Function<Row, Student>() {
            @Override
            public Student call(Row row) throws Exception {
                Student stu = new Student();
                stu.setAge(row.getInt(0));
                stu.setID(row.getInt(1));
                stu.setName(row.getString(2));
                return stu;
            }
        });
        List<Student> studentList = teengerStudentRDD.collect();
        for (Student stu : studentList){
            System.out.println(stu);
        }
    }
}

测试：

最后

以上就是愤怒大山最近收集整理的关于使用反射的方式将RDD转换为DataFrame的全部内容，更多相关使用反射内容请搜索靠谱客的其他文章。