java版

    1. /**
    2.  * 以编程方式动态指定元数据，将RDD转换为DataFrame
    3.  * @author Administrator
    4.  *
    5.  */
    6. public class RDD2DataFrameProgrammatically {
    8.     public static void main(String[] args) {
    9.         // 创建SparkConf、JavaSparkContext、SQLContext
    10.         SparkConf conf = new SparkConf()
    11.                 .setMaster("local")  
    12.                 .setAppName("RDD2DataFrameProgrammatically");  
    13.         JavaSparkContext sc = new JavaSparkContext(conf);
    14.         SQLContext sqlContext = new SQLContext(sc);
    16.         // 第一步，创建一个普通的RDD，但是，必须将其转换为RDD<Row>的这种格式
    17.         JavaRDD<String> lines = sc.textFile("C://Users//Administrator//Desktop//students.txt");
    19.         // 分析一下
    20.         // 它报了一个，不能直接从String转换为Integer的一个类型转换的错误
    21.         // 就说明什么，说明有个数据，给定义成了String类型，结果使用的时候，要用Integer类型来使用
    22.         // 而且，错误报在sql相关的代码中
    23.         // 所以，基本可以断定，就是说，在sql中，用到age<=18的语法，所以就强行就将age转换为Integer来使用
    24.         // 但是，肯定是之前有些步骤，将age定义为了String
    25.         // 所以就往前找，就找到了这里
    26.         // 往Row中塞数据的时候，要注意，什么格式的数据，就用什么格式转换一下，再塞进去
    27.         JavaRDD<Row> studentRDD = lines.map(new Function<String, Row>() {
    29.             private static final long serialVersionUID = 1L;
    31.             @Override
    32.             public Row call(String line) throws Exception {
    33.                 String[] lineSplited = line.split(","); 
    34.                 return RowFactory.create(
    35.                         Integer.valueOf(lineSplited[0]), 
    36.                         lineSplited[1], 
    37.                         Integer.valueOf(lineSplited[2]));      
    38.             }
    40.         });
    42.         // 第二步，动态构造元数据
    43.         // 比如说，id、name等，field的名称和类型，可能都是在程序运行过程中，动态从mysql db里
    44.         // 或者是配置文件中，加载出来的，是不固定的
    45.         // 所以特别适合用这种编程的方式，来构造元数据
    46.         List<StructField> structFields = new ArrayList<StructField>();
    47.         structFields.add(DataTypes.createStructField("id", DataTypes.IntegerType, true));  
    48.         structFields.add(DataTypes.createStructField("name", DataTypes.StringType, true));  
    49.         structFields.add(DataTypes.createStructField("age", DataTypes.IntegerType, true));  
    50.         StructType structType = DataTypes.createStructType(structFields);
    52.         // 第三步，使用动态构造的元数据，将RDD转换为DataFrame
    53.         DataFrame studentDF = sqlContext.createDataFrame(studentRDD, structType);
    55.         // 后面，就可以使用DataFrame了
    56.         studentDF.registerTempTable("students");  
    58.         DataFrame teenagerDF = sqlContext.sql("select * from students where age<=18");  
    60.         List<Row> rows = teenagerDF.javaRDD().collect();
    61.         for(Row row : rows) {
    62.             System.out.println(row);  
    63.         }
    64.     }
    66. }

    scala 版

    1. /**
    2.  * @author Administrator
    3.  */
    4. object RDD2DataFrameProgrammatically extends App {
    6.   val conf = new SparkConf()
    7.       .setMaster("local")  
    8.       .setAppName("RDD2DataFrameProgrammatically")  
    9.   val sc = new SparkContext(conf)
    10.   val sqlContext = new SQLContext(sc)
    12.   // 第一步，构造出元素为Row的普通RDD
    13.   val studentRDD = sc.textFile("C://Users//Administrator//Desktop//students.txt", 1)
    14.       .map { line => Row(line.split(",")(0).toInt, line.split(",")(1), line.split(",")(2).toInt) } 
    16.   // 第二步，编程方式动态构造元数据
    17.   val structType = StructType(Array(
    18.       StructField("id", IntegerType, true),
    19.       StructField("name", StringType, true),
    20.       StructField("age", IntegerType, true)))  
    22.   // 第三步，进行RDD到DataFrame的转换
    23.   val studentDF = sqlContext.createDataFrame(studentRDD, structType)  
    25.   // 继续正常使用
    26.   studentDF.registerTempTable("students")  
    28.   val teenagerDF = sqlContext.sql("select * from students where age<=18")  
    30.   val teenagerRDD = teenagerDF.rdd.collect().foreach { row => println(row) }    
    31. }