标签:
一、通过结构化数据创建DataFrame:
.setAppName("DataFrameCreate").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);
        
        DataFrame df = sqlContext.read().json("hdfs://spark1:9000/students.json");  //结构化数据直接加载为DataFrame
        
        df.show();  
    }
二、通过RDD创建DataFrame的两种创建方式
 (数据源students.txt的数据截图)
 (数据源students.txt的数据截图)
2.1通过已知类型的schema创建DataFrame,代码如下:
2.2手动创建schema的方式创建DataFrame
public static void main(String[] args) {
	JavaRDD<String> lines = sc.textFile("D://students.txt");
        
        //将普通RDD装换成JavaRDD<Row>
        JavaRDD<Row> rowRDD = lines.map(new Function<String, Row>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Row call(String line) throws Exception {
                String[] strArray = line.split(",");
                Row row= RowFactory.create(
                        Integer.valueOf(strArray[0]),    //id
                        strArray[1],    //name
                        Integer.valueOf(strArray[2]));    //age
            
                return row;
            }
        });
        
        //第二步 创建元类型, 即创建schema
        List<StructField> structFields = new ArrayList<StructField>();
        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 structType = DataTypes.createStructType(structFields);
        //根据元数据类型将JavaRDD<Row>转化成DataFrame
        DataFrame studentDF = sqlCotnext.createDataFrame(rowRDD, structType);
        studentDF.show();
    }
-》DataFrame、RDD、List互转
List<Row> studentList = rows.collect();
三、DataFrame基本用法
	df.show();
        // 打印DataFrame的元数据(Schema)
        df.printSchema();
        // 查询某列所有的数据
        df.select("name").show();  
        // 查询某几列所有的数据,并对列进行计算
        df.select(df.col("name"), df.col("age").plus(1)).show();
        // 根据某一列的值进行过滤
        df.filter(df.col("age").gt(18)).show();
        // 根据某一列进行分组,然后进行聚合
        df.groupBy(df.col("age")).count().show();
标签:
原文地址:http://www.cnblogs.com/key1309/p/5352310.html