zoukankan      html  css  js  c++  java
  • 关于sparksql

    1.读取json文件,并且进行查询等操作

    所使用的jar包为

    json文件内容

    { "id":1 ,"name":" Ella","age":36 } { "id":2,"name":"Bob","age":29 } { "id":3 ,"name":"Jack","age":29 } { "id":4 ,"name":"Jim","age":28 } { "id":5 ,"name":"Damon" } { "id":5 ,"name":"Damon" }

      val conf = new SparkConf().setAppName("DataFrameTest").setMaster("local")
        val sc = new SparkContext(conf)
        val sqlContext = new SQLContext(sc)  
        val df = sqlContext.jsonFile("H:\文件\数据集\test1\1.json");
       df.show()       //查询所有 df.distinct.show() //去重
    df.filter(df.col("age")>20).show() //age>20的行 df.groupBy("name").count().show() //根据name分组 df.sort(df("name").asc).show() //将数据按 name 升序排列 df.head(3).foreach(print) //取出前 3 行数据 df.select(df("name").as("username")).show() //查询所有记录的 name 列,并为其取别名为 username df.agg("age"->"avg").foreach(print) //查询年龄 age 的平均值 df.agg("age"->"min").foreach(print) //) 查询年龄 age 的最小值

    2.编程实现将 RDD 转换为 DataFrame

    文件内容

    1,Ella,36

    2,Bob,29

    3,Jack,29

     val conf = new SparkConf()  
           conf.setMaster("local")  
               .setAppName("Testsql")  
           val sc = new SparkContext(conf)  
           val sqlContext = new SQLContext(sc)
           //hdfs://192.168.6.134:9000/wys/1.txt
           //H:文件数据集
           val aRDD = sc.textFile("H:\文件\数据集\test1\2.txt", 1)
          .map { line => Row(line.split(",")(0), line.split(",")(1),line.split(",")(2) )} 
      
          // 第二步,编程方式动态构造元数据
          val structType = StructType(Array(
              StructField("id", StringType, true),
              StructField("name", StringType, true),
              StructField("age", StringType, true)))  
          
          // 第三步,进行RDD到DataFrame的转换
          val aDF = sqlContext.createDataFrame(aRDD, structType)  
          
          // 继续正常使用
          aDF.registerTempTable("A")  
          val teenagerDF4 = sqlContext.sql("select id,name,age from A")  
          teenagerDF4.map(t => "id:"+t(0)+","+"name:"+t(1)+","+"age:"+t(2)).foreach(println)

    3.编程实现利用 DataFrame 读写 MySQL 的数据

     val conf = new SparkConf()  
           conf.setMaster("local")  
                 .setAppName("Testsql")  
          val sc = new SparkContext(conf) 
          val sqlContext = new SQLContext(sc)
          val employeeRDD = sc.parallelize(Array("3 Mary F 26","4 Tom M 23")).map(_.split(" "))
          val schema = StructType(List(StructField("id", IntegerType,true),StructField("name", StringType, true),StructField("gender", StringType,true),StructField("age", IntegerType, true)))
          val rowRDD = employeeRDD.map(p => Row(p(0).toInt,p(1).trim,p(2).trim,p(3).toInt))
          val employeeDF = sqlContext.createDataFrame(rowRDD, schema)
          val prop = new Properties()
          prop.put("user", "root")
          prop.put("password", "root")
          prop.put("driver","com.mysql.jdbc.Driver")
          employeeDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/sparktest","sparktest.spark", prop)
          val jdbcDF = sqlContext.read.format("jdbc").option("url","jdbc:mysql://localhost:3306/sparktest").option("driver","com.mysql.jdbc.Driver").option("dbtable","spark").option("user","root").option("password", "root").load()
          jdbcDF.agg("age" -> "max", "age" -> "sum")
  • 相关阅读:
    H5 移动端相册拍照 录音 录像,然后上传后台
    h5 移动端调用相机 摄像机 录音机获取文件,并下载到本地
    Java 判断字符串是否含有 数字 字母 特殊字符
    java 以0开头的数字计算,保留开头0
    Spring 与hibernate 整合,测试数据连接
    http://blog.csdn.net/basycia/article/details/52205916
    MySQL数据库基础知识002
    数据库基础知识001
    数组排序
    输出杨辉三角
  • 原文地址:https://www.cnblogs.com/wys-373/p/10602919.html
Copyright © 2011-2022 走看看