关于sparksql
2024-09-05 13:59:52
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")>).show() //age>20的行
df.groupBy("name").count().show() //根据name分组
df.sort(df("name").asc).show() //将数据按 name 升序排列
df.head().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", )
.map { line => Row(line.split(",")(), line.split(",")(),line.split(",")() )} // 第二步,编程方式动态构造元数据
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()+","+"name:"+t()+","+"age:"+t()).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().toInt,p().trim,p().trim,p().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")
最新文章
- CentOS 7.2安装docker-compose运行gitlib
- byte为什么要与上0xff?
- Visual studio 2013安装
- Interface/接口
- 多媒体(4):JPEG图像压缩编码
- HDU 3966(树链剖分+点修改+点查询)
- [SAP ABAP开发技术总结]BAPI调用
- 《ArcGIS Engine+C#实例开发教程》第五讲 鹰眼的实现
- 【HDOJ】1076 An Easy Task
- Java 反射 Array动态创建数组
- hive:数据库“行专列”操作---使用collect_set/collect_list/collect_all &; row_number()over(partition by 分组字段 [order by 排序字段])
- java -jar参数携带问题
- clone内容包含select2
- Python中的基本数据类型的区别
- 安全SOCKET
- mongodb(五)
- 图解linux安装tomcat(附常用命令)
- select和epoll概念
- 想进BAT?这些面试题助你一臂之力
- UDP网络程序实例