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")