【Spark-SQL学习之二】 SparkSQL DataFrame创建和储存

package com.wjy.df;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.rdd.RDD;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

/**
 * 读取json格式的文件创建DataFrame
 * 
 * 注意 ：json文件中不能嵌套json格式的内容
 * 
 * 1.读取json格式两种方式
 * 2.df.show默认显示前20行，使用df.show(行数)显示多行
 * 3.df.javaRDD/(scala df.rdd) 将DataFrame转换成RDD
 * 4.df.printSchema()显示DataFrame中的Schema信息
 * 5.dataFram自带的API 操作DataFrame ，用的少
 * 6.想使用sql查询，首先要将DataFrame注册成临时表：df.registerTempTable("jtable")，再使用sql,怎么使用sql?sqlContext.sql("sql语句")
 * 7.不能读取嵌套的json文件
 * 8.df加载过来之后将列按照ascii排序了
 * @author root
 *
 */
public class CreateDFFromJosonFile {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("CreateDFFromJosonFile");
        SparkContext sc = new SparkContext(conf);//注意 这里不是JavaSparkContext
        //创建SQLContext
        SQLContext sqlContext = new SQLContext(sc);
        
        /**
         * DataFrame的底层是一个一个的RDD  RDD的泛型是Row类型。
         * 以下两种方式都可以读取json格式的文件
         * {"name":"zhangsan","age":"20"}
           {"name":"lisi"}
           {"name":"wangwu","age":"18"}
         */
        DataFrame df = sqlContext.read().format("json").load("./data/json");//{"name":"zhangsan","age":"20"};
        df.show();// 显示 DataFrame中的内容，默认显示前20行。如果现实多行要指定多少行show(行数) 注意：当有多个列时，显示的列先后顺序是按列的ascii码先后显示。
        DataFrame df2 = sqlContext.read().json("./data/json");
        df2.show();
        /*
         * +----+--------+
           | age|    name|
           +----+--------+
           |  20|zhangsan|
           |null|    lisi|
           |  18|  wangwu|
           +----+--------+
         */
        
        //DataFrame转换成RDD
        JavaRDD<Row> javaRDD = df.javaRDD();
        //树形的形式显示schema信息
        df.printSchema();
        /*
         * root
             |-- age: string (nullable = true)
             |-- name: string (nullable = true)    
         */
        
        //dataFram自带的API 操作DataFrame 这种方式比较麻烦 用的比较少
        //select name from table
        df.select("name").show();
        /*
         * +--------+
           |    name|
           +--------+
           |zhangsan|
           |    lisi|
           |  wangwu|
           +--------+
         */
        //select name ,age+10 as addage from table
        df.select(df.col("name"),df.col("age").plus(10).alias("addage")).show();
        /*
         * +--------+------+
           |    name|addage|
           +--------+------+
           |zhangsan|  30.0|
           |    lisi|  null|
           |  wangwu|  28.0|
           +--------+------+
         */
        //select name ,age from table where age>19
        df.select(df.col("name"),df.col("age")).where(df.col("age").gt(19)).show();
        /*
         * +--------+---+
           |    name|age|
           +--------+---+
           |zhangsan| 20|
           +--------+---+
         */
        //select age,count(*) from table group by age
        df.groupBy(df.col("age")).count().show();
        /*
         * +----+-----+
           | age|count|
           +----+-----+
           |  18|    1|
           |  20|    1|
           |null|    1|
           +----+-----+
         */
        
        //将DataFrame注册成临时的一张表，这张表相当于临时注册到内存中，是逻辑上的表，不会物化到磁盘  这种方式用的比较多
        df.registerTempTable("person");
        DataFrame df3 = sqlContext.sql("select age,count(*) as gg from person group by age");
        df3.show();
        DataFrame df4 = sqlContext.sql("select age, name from person");
        df4.show();
        
        sc.stop();
    }

}

package com.wjy.df

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext

object CreateDFFromJsonFile {
  def main(args:Array[String]):Unit={
    val conf = new SparkConf().setMaster("local").setAppName("CreateDFFromJsonFile");
    val sc = new SparkContext(conf);
    val sqlContext = new SQLContext(sc);
    
    val df1 = sqlContext.read.json("./data/json");
    df1.show();
    val df2 = sqlContext.read.format("json").load("./data/json");
    df2.show();
    
    val rdd = df1.rdd;
    df1.printSchema();
    
    //select name from table
    df1.select(df1.col("name")).show();
    //select name from table where age>19
    df1.select(df1.col("name"),df1.col("age")).where(df1.col("age").gt(19)).show();
    //select count(*) from table group by age
    df1.groupBy(df1.col("age")).count().show();
    
    //注册临时表
    df1.registerTempTable("person");
    val df3 = sqlContext.sql("select * from person");
    df3.show();
    /*
     * +----+--------+
             | age|    name|
       +----+--------+
       |  20|zhangsan|
       |null|    lisi|
       |  18|  wangwu|
       +----+--------+
     */
    sc.stop();
  }
}

./spark-submit 
--master spark://node1:7077,node2:7077 
--executor-cores 1 
--executor-memory 2G 
--total-executor-cores 1
--class com.bjsxt.sparksql.dataframe.CreateDFFromHive 
/root/test/HiveTest.jar

<configuration>
<property>
<name>hive.metastore.uris</name>
<value>thrift://node1:9083</value>
</property>
</configuration>

./spark-shell 
--master spark://node1:7077,node2:7077 
--executor-cores 1 
--executor-memory 1g 
--total-executor-cores 1

......

scala>import org.apache.spark.sql.hive.HiveContext;
scala>val hc = new HiveContext(sc);
scala>hc.sql("show databases").show();
scala>hc.sql("user default").show();
scala>hc.sql("select count(*) from jizhan").show();

export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop