sparlsql有哪些

这篇文章给大家分享的是有关sparl sql有哪些的内容。小编觉得挺实用的，因此分享给大家做个参考，一起跟随小编过来看看吧。

为永丰等地区用户提供了全套网页设计制作服务，及永丰网站建设行业解决方案。主营业务为网站设计、成都网站制作、永丰网站设计，以传统方式定制建设网站，并提供域名空间备案等一条龙服务，秉承以专业、用心的态度为用户提供真诚的服务。我们深信只要达到每一位用户的要求，就会得到认可，从而选择与我们长期合作。这样，我们也可以走得更远！

1、读取json格式的文件创建DataFrame

java (spark1.6)

public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setMaster("local").setAppName("javaSpark01");
        SparkContext sc = new SparkContext(conf);

        SQLContext sqlContext = new SQLContext(sc);
//        Dataset df = sqlContext.read().format("json").load("G:/idea/scala/spark02/json");
        Dataset df2 = sqlContext.read().json("G:/idea/scala/spark02/json");
        df2.show();

        //树形的形式显示schema信息
        df2.printSchema();

        //注册临时表 将DataFrame注册成临时的一张表，这张表临时注册到内存中，是逻辑上的表，不会雾化到磁盘
        df2.registerTempTable("baidukt_table");
        Dataset sql = sqlContext.sql("select * from baidukt_table");
        sql.show();
        Dataset sql1 = sqlContext.sql("select age,count(1) from baidukt_table group by age");
        sql1.show();
 }

scala(spark 1.6)

def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local").setAppName("Spark08 1.6")
     val sc = new SparkContext(conf)
    val sqlContext: SQLContext = new SQLContext(sc)

    val df = sqlContext.read.format("json").load("G:/idea/scala/spark02/json")
//    val df1 = sqlContext.read.json("G:/idea/scala/spark02/json")
    //显示前50行数据
    df.show(50)
    //树形的形式显示schema信息
    df.printSchema()
    //注册临时表
    df.registerTempTable("baidukt_com_table")
    val result = sqlContext.sql("select age,count(1) from baidukt_com_table group by age")
    result.show()
    val result1 = sqlContext.sql("select * from baidukt_com_table")
    result1.show()
    sc.stop()
  }

java (spark 2.0++)

public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("Spark 2.0 ++");
        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
        Dataset df = spark.read().json("G:/idea/scala/spark02/json");
//        Dataset df1 = spark.read().format("json").load("G:/idea/scala/spark02/json");
        df.show();
        df.printSchema();

        df.createOrReplaceGlobalTempView("baidu_com_spark2");
        Dataset resut = spark.sql("select * from baidu_com_spark2");
        resut.show();
        spark.stop();
    }

scala(spark 2.0++)

  def main(args: Array[String]): Unit = {
    //用户的当前工作目录
//    val  location = System.setProperties("user.dir","spark_2.0"
    val conf = new SparkConf().setAppName("Spark08 2.0++").setMaster("local[3]")
    val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()

    //数据导入方式
    val df: DataFrame = spark.read.json("G:/idea/scala/spark02/json")
//    val df1: DataFrame = spark.read.format("json").load("G:/idea/scala/spark02/json")
    //查看表
    df.show()
    //查看表
    df.printSchema()

    //直接使用spark SQL进行查询
    //先注册为临时表
    //createOrReplaceTempView：创建临时视图，此视图的生命周期与用于创建此数据集的[SparkSession]相关联。
    //createGlobalTempView：创建全局临时视图，此时图的生命周期与Spark Application绑定。
    df.createOrReplaceTempView("people")
    val result: DataFrame = spark.sql("select * from people")
    result.show()
    spark.stop()
  }

2、通过json格式的RDD创建DataFrame

java(spark 1.6)

public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setMaster("local").setAppName("jsonRDD");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);
        JavaRDD data = sc.parallelize(Arrays.asList(
                "{\"name\":\"zhangsan\",\"score\":\"100\"}",
                "{\"name\":\"lisi\",\"score\":\"200\"}",
                "{\"name\":\"wangwu\",\"score\":\"300\"}"
        ));
        Dataset df = sqlContext.read().json(data);
        df.show();
        df.printSchema();

        df.createOrReplaceTempView("baidu_com_spark2");
        Dataset resut = sqlContext.sql("select * from baidu_com_spark2");
        resut.show();
        sc.stop();
    }

scala(spark 1.6)

def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("spark10")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    val data: RDD[String] = sc.parallelize(Array(
      "{\"name\":\"zhangsan\",\"age\":18}",
      "{\"name\":\"lisi\",\"age\":19}",
      "{\"name\":\"wangwu\",\"age\":20}"
    ))
    val df = sqlContext.read.json(data)
    df.show()
    df.printSchema()
    df.createOrReplaceTempView("baidukt_com_spark1.6")
    val result = sqlContext.sql("select * from baidukt_com_spark1.6")
    result.show()
    result.printSchema()
    sc.stop()
  }

3、非json格式的RDD创建DataFrame

3.1 通过反射的方式将非json格式的RDD转换成DataFrame（不推荐，所以不复制代码过来了）

3.2、态创建Schema将非json格式的RDD转换成DataFrame

4、读取parquet文件创建DataFrame（多次io 不推荐）

5、读取JDBC中的数据创建DataFrame(MySQL为例)

java(spark 1.6)

   public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setMaster("local").setAppName("mysql");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);
        /**
         * 第一种方式读取MySql数据库表，加载为DataFrame
         */
        Map options = new HashMap();
        options.put("url", "jdbc:mysql://localhost:3306/spark");//连接地址和数据库名称
        options.put("driver", "com.mysql.jdbc.Driver");//驱动
        options.put("user", "root");//用户名
        options.put("password", "admin");//密码
        options.put("dbtable", "person");//表
        Dataset person = sqlContext.read().format("jdbc").options(options).load();
        person.show();
        //注册临时表
        person.registerTempTable("person");
        /**
         * 第二种方式读取MySql数据表加载为DataFrame
         */
        DataFrameReader reader = sqlContext.read().format("jdbc");
        reader.option("url", "jdbc:mysql://localhost:3306/spark");
        reader.option("driver", "com.mysql.jdbc.Driver");
        reader.option("user", "root");
        reader.option("password", "admin");
        reader.option("dbtable", "score");
        Dataset score = reader.load();
        score.show();
        score.registerTempTable("score");

        Dataset result = sqlContext.sql("select person.id,person.name,score.score from person,score where person.name = score.name");
        result.show();
        /**
         * 将DataFrame结果保存到Mysql中
         */
        Properties properties = new Properties();
        properties.setProperty("user", "root");
        properties.setProperty("password", "admin");
        result.write().mode(SaveMode.Overwrite).jdbc("jdbc:mysql://localhost:3306/spark", "result", properties);
        sc.stop();
    }

scala (spark 1.6)

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local").setAppName("mysql")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    /**
      * 第一种方式读取Mysql数据库表创建DF
      */
    val options = new mutable.HashMap[String,String]();
    options.put("url", "jdbc:mysql://localhost:3306/spark")
    options.put("driver","com.mysql.jdbc.Driver")
    options.put("user","root")
    options.put("password", "admin")
    options.put("dbtable","person")
    val person = sqlContext.read.format("jdbc").options(options).load()
    person.show()
    person.registerTempTable("person")
    /**
      * 第二种方式读取Mysql数据库表创建DF
      */
    val reader = sqlContext.read.format("jdbc")
    reader.option("url", "jdbc:mysql://localhost:3306/spark")
    reader.option("driver","com.mysql.jdbc.Driver")
    reader.option("user","root")
    reader.option("password","admin")
    reader.option("dbtable", "score")
    val score = reader.load()
    score.show()
    score.registerTempTable("score")
    val result = sqlContext.sql("select person.id,person.name,score.score from person,score where person.name = score.name")
    result.show()
    /**
      * 将数据写入到Mysql表中
      */
    val properties = new Properties()
    properties.setProperty("user", "root")
    properties.setProperty("password", "admin")
    result.write.mode(SaveMode.Append).jdbc("jdbc:mysql://localhost:3306/spark", "result", properties)
    sc.stop()
  }

6、读取Hive中的数据加载成DataFrame

HiveContext是SQLContext的子类，连接Hive建议使用HiveContext。

java(spark 1.6)

public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("hive");
        JavaSparkContext sc = new JavaSparkContext(conf);
        //HiveContext是SQLContext的子类。
        HiveContext hiveContext = new HiveContext(sc);
        hiveContext.sql("USE spark");
        hiveContext.sql("DROP TABLE IF EXISTS student_infos");
        //在hive中创建student_infos表
        hiveContext.sql("CREATE TABLE IF NOT EXISTS student_infos (name STRING,age INT) row format delimited fields terminated by '\t' ");
        hiveContext.sql("load data local inpath '/root/test/student_infos' into table student_infos");

        hiveContext.sql("DROP TABLE IF EXISTS student_scores");
        hiveContext.sql("CREATE TABLE IF NOT EXISTS student_scores (name STRING, score INT) row format delimited fields terminated by '\t'");
        hiveContext.sql("LOAD DATA LOCAL INPATH '/root/test/student_scores INTO TABLE student_scores");
        /**
         * 查询表生成DataFrame
         */
        Dataset goodStudentsDF = hiveContext.sql("SELECT si.name, si.age, ss.score FROM student_infos si JOIN student_scores ss ON si.name=ss.name WHERE ss.score>=80");
        hiveContext.sql("DROP TABLE IF EXISTS good_student_infos");
        goodStudentsDF.registerTempTable("goodstudent");
        Dataset result = hiveContext.sql("select * from goodstudent");
        result.show();
        result.printSchema();

        /**
         * 将结果保存到hive表 good_student_infos
         */
        goodStudentsDF.write().mode(SaveMode.Overwrite).saveAsTable("good_student_infos");

        Row[] goodStudentRows = hiveContext.table("good_student_infos").collect();
        for(Row goodStudentRow : goodStudentRows) {
            System.out.println(goodStudentRow);
        }
        sc.stop();
    }

scala(spark 1.6)

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("HiveSource")
    val sc = new SparkContext(conf)
    /**
      * HiveContext是SQLContext的子类。
      */
    val hiveContext = new HiveContext(sc)
    hiveContext.sql("use spark")
    hiveContext.sql("drop table if exists student_infos")
    hiveContext.sql("create table if not exists student_infos (name string,age int) row format  delimited fields terminated by '\t'")
    hiveContext.sql("load data local inpath '/root/test/student_infos' into table student_infos")

    hiveContext.sql("drop table if exists student_scores")
    hiveContext.sql("create table if not exists student_scores (name string,score int) row format delimited fields terminated by '\t'")
    hiveContext.sql("load data local inpath '/root/test/student_scores' into table student_scores")

    val df = hiveContext.sql("select si.name,si.age,ss.score from student_infos si,student_scores ss where si.name = ss.name")
    hiveContext.sql("drop table if exists good_student_infos")
    /**
      * 将结果写入到hive表中
      */
    df.write.mode(SaveMode.Overwrite).saveAsTable("good_student_infos")
    sc.stop()
  }

7、自定义udf

scala(spark 1.6)

 def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("spark13")
    val spark = SparkSession.builder().config(conf).getOrCreate()
    //rdd转df
    val rdd: RDD[String] = spark.sparkContext.parallelize(Array("zhangsan","wangwu","masi"))
    val rowRDD: RDD[Row] = rdd.map(RowFactory.create(_))
    val schema = DataTypes.createStructType(Array(StructField("name",StringType,true)))
    val df: DataFrame = spark.sqlContext.createDataFrame(rowRDD,schema)

    df.show(50)
    
    df.printSchema()

    df.createOrReplaceTempView("test")
    //自定义udf函数 函数名为StrLen，参数为String、Int String有问题，网上说需要java.lang.String类型 
//    spark.sqlContext.udf.register("StrLen",(s:String,i:Int)=>{s.length+i})
//    spark.sqlContext.udf.register("StrLen",(i:Int)=>{i})
//    spark.sql("select name ,StrLen(name,10) as length from test").show(20)
    spark.sql("select name ,StrLen(10) as length from test").show(20)
  }

java (spark 1.6)

  public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("udf");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);
        JavaRDD parallelize = sc.parallelize(Arrays.asList("zhansan","lisi","wangwu"));
        JavaRDD rowRDD = parallelize.map(new Function() {

            private static final long serialVersionUID = 1L;

            @Override
            public Row call(String s) throws Exception {
                return RowFactory.create(s);
            }
        });

        List fields = new ArrayList();
        fields.add(DataTypes.createStructField("name", DataTypes.StringType,true));

        StructType schema = DataTypes.createStructType(fields);
        Dataset df = sqlContext.createDataFrame(rowRDD, schema);
        df.registerTempTable("user");

        /**
         * 根据UDF函数参数的个数来决定是实现哪一个UDF  UDF1，UDF2。。。。UDF1xxx
         */
        sqlContext.udf().register("StrLen", new UDF1() {

            private static final long serialVersionUID = 1L;

            @Override
            public Integer call(String t1) throws Exception {
                return t1.length();
            }
        }, DataTypes.IntegerType);
        sqlContext.sql("select name ,StrLen(name) as length from user").show();
        sc.stop();
    }

感谢各位的阅读！关于“sparl sql有哪些”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，让大家可以学到更多知识，如果觉得文章不错，可以把它分享出去让更多的人看到吧！