JSON数据源

  1. Spark SQL可以自动推断JSON文件的元数据，并且加载其数据，创建一个DataFrame。可以使用SQLContext.read.json()方法，针对一个元素类型为String的RDD，或者是一个JSON文件。
  3. 但是要注意的是，这里使用的JSON文件与传统意义上的JSON文件是不一样的。每行都必须，也只能包含一个，单独的，自包含的，有效的JSON对象。不能让一个JSON对象分散在多行。否则会报错。
  5. 综合性复杂案例：查询成绩为80分以上的学生的基本信息与成绩信息

案例

  1. /**
  2.  * JSON数据源
  3.  * @author Administrator
  4.  *
  5.  */
  6. public class JSONDataSource {
  8.     public static void main(String[] args) {
  9.         SparkConf conf = new SparkConf()
  10.                 .setAppName("JSONDataSource");  
  11.         JavaSparkContext sc = new JavaSparkContext(conf);
  12.         SQLContext sqlContext = new SQLContext(sc);
  14.         // 针对json文件，创建DataFrame（针对json文件创建DataFrame）
  15.         DataFrame studentScoresDF = sqlContext.read().json(
  16.                 "hdfs://spark1:9000/spark-study/students.json");  
  18.         // 针对学生成绩信息的DataFrame，注册临时表，查询分数大于80分的学生的姓名
  19.         // （注册临时表，针对临时表执行sql语句）
  20.         studentScoresDF.registerTempTable("student_scores");
  21.         DataFrame goodStudentScoresDF = sqlContext.sql(
  22.                 "select name,score from student_scores where score>=80");
  24.         // （将DataFrame转换为rdd，执行transformation操作）
  25.         List<String> goodStudentNames = goodStudentScoresDF.javaRDD().map(
  27.                 new Function<Row, String>() {
  29.                     private static final long serialVersionUID = 1L;
  31.                     @Override
  32.                     public String call(Row row) throws Exception {
  33.                         return row.getString(0);
  34.                     }
  36.                 }).collect();
  38.         // 然后针对JavaRDD<String>，创建DataFrame
  39.         // （针对包含json串的JavaRDD，创建DataFrame）
  40.         List<String> studentInfoJSONs = new ArrayList<String>();
  41.         studentInfoJSONs.add("{\"name\":\"Leo\", \"age\":18}");  
  42.         studentInfoJSONs.add("{\"name\":\"Marry\", \"age\":17}");  
  43.         studentInfoJSONs.add("{\"name\":\"Jack\", \"age\":19}");
  44.         JavaRDD<String> studentInfoJSONsRDD = sc.parallelize(studentInfoJSONs);
  45.         DataFrame studentInfosDF = sqlContext.read().json(studentInfoJSONsRDD);
  47.         // 针对学生基本信息DataFrame，注册临时表，然后查询分数大于80分的学生的基本信息
  48.         studentInfosDF.registerTempTable("student_infos");  
  50.         String sql = "select name,age from student_infos where name in (";
  51.         for(int i = 0; i < goodStudentNames.size(); i++) {
  52.             sql += "'" + goodStudentNames.get(i) + "'";
  53.             if(i < goodStudentNames.size() - 1) {
  54.                 sql += ",";
  55.             }
  56.         }
  57.         sql += ")";
  59.         DataFrame goodStudentInfosDF = sqlContext.sql(sql);
  61.         // 然后将两份数据的DataFrame，转换为JavaPairRDD，执行join transformation
  62.         // （将DataFrame转换为JavaRDD，再map为JavaPairRDD，然后进行join）
  63.         JavaPairRDD<String, Tuple2<Integer, Integer>> goodStudentsRDD = 
  65.                 goodStudentScoresDF.javaRDD().mapToPair(new PairFunction<Row, String, Integer>() {
  67.                     private static final long serialVersionUID = 1L;
  69.                     @Override
  70.                     public Tuple2<String, Integer> call(Row row) throws Exception {
  71.                         return new Tuple2<String, Integer>(row.getString(0), 
  72.                                 Integer.valueOf(String.valueOf(row.getLong(1))));  
  73.                     }
  75.                 }).join(goodStudentInfosDF.javaRDD().mapToPair(new PairFunction<Row, String, Integer>() {
  77.                     private static final long serialVersionUID = 1L;
  79.                     @Override
  80.                     public Tuple2<String, Integer> call(Row row) throws Exception {
  81.                         return new Tuple2<String, Integer>(row.getString(0),
  82.                                 Integer.valueOf(String.valueOf(row.getLong(1))));   
  83.                     }
  85.                 }));
  87.         // 然后将封装在RDD中的好学生的全部信息，转换为一个JavaRDD<Row>的格式
  88.         // （将JavaRDD，转换为DataFrame）
  89.         JavaRDD<Row> goodStudentRowsRDD = goodStudentsRDD.map(
  91.                 new Function<Tuple2<String,Tuple2<Integer,Integer>>, Row>() {
  93.                     private static final long serialVersionUID = 1L;
  95.                     @Override
  96.                     public Row call(
  97.                             Tuple2<String, Tuple2<Integer, Integer>> tuple)
  98.                             throws Exception {
  99.                         return RowFactory.create(tuple._1, tuple._2._1, tuple._2._2);
  100.                     }
  102.                 });
  104.         // 创建一份元数据，将JavaRDD<Row>转换为DataFrame
  105.         List<StructField> structFields = new ArrayList<StructField>();
  106.         structFields.add(DataTypes.createStructField("name", DataTypes.StringType, true)); 
  107.         structFields.add(DataTypes.createStructField("score", DataTypes.IntegerType, true));  
  108.         structFields.add(DataTypes.createStructField("age", DataTypes.IntegerType, true));  
  109.         StructType structType = DataTypes.createStructType(structFields);
  111.         DataFrame goodStudentsDF = sqlContext.createDataFrame(goodStudentRowsRDD, structType);
  113.         // 将好学生的全部信息保存到一个json文件中去
  114.         // （将DataFrame中的数据保存到外部的json文件中去）
  115.         goodStudentsDF.write().format("json").save("hdfs://spark1:9000/spark-study/good-students");  
  116.     }
  118. }