首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >SparkSQL常用操作

SparkSQL常用操作

作者头像
UFO
发布2018-09-04 11:50:16
4380
发布2018-09-04 11:50:16
举报
文章被收录于专栏:Spark生态圈Spark生态圈

 1、从json文件创建dataFrame

val df: DataFrame = sqlContext.read.json("hdfs://master:9000/user/spark/data/people.json") val people = df.registerTempTable("person") val teenegers: DataFrame = sqlContext.sql("select name,age from person") teenegers.map(x => "name:" + x(0)+ " " + "age:" + x(1)).collect().foreach(println)

2、从parquet文件创建dataFrame

val df: DataFrame = sqlContext.read.parquet("hdfs://master:9000/user/spark/data/namesAndAges.parquet") val people = df.registerTempTable("person") val teenegers: DataFrame = sqlContext.sql("select name,age from person") teenegers.map(x => "name:" + x(0)+ " " + "age:" + x(1)).collect().foreach(println)

3、从普通RDD创建dataFrame_1

val people = sc.textFile("hdfs://master:9000/user/spark/data/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF people.registerTempTable("people") val teenagers = sqlContext.sql("select name,age from people") teenagers.map(x => "name:" + x(0)+ " " + "age:" + x(1)).collect().foreach(println)

4、从普通RDD创建dataFrame_2

val people = sc.textFile("hdfs://master:9000/user/spark/data/people.txt") val schemaString = "name age" import org.apache.spark.sql.Row import org.apache.spark.sql.types.{StructType,StructField,StringType} val schema = StructType(schemaString.split(" ").map(fieldName => StructField(fieldName,StringType,true))) val rowRDD = people.map(_.split(",")).map(x => Row(x(0),x(1).trim)) val df: DataFrame = sqlContext.createDataFrame(rowRDD,schema) df.registerTempTable("people")val teenagers = sqlContext.sql("select name,age from people") teenagers.map(x => "name:" + x(0)+ " " + "age:" + x(1)).collect().foreach(println)

5、测试dataframe的read和save方法(注意load方法默认是加载parquet文件)

val df = sqlContext.read.load("hdfs://master:9000/user/spark/data/namesAndAges.parquet") df.select("name").write.save("hdfs://master:9000/user/spark/data/name.parquet")

6、测试dataframe的read和save方法(可通过手动设置数据源和保存测mode)

val df =sqlContext.read.format("json").load("hdfs://master:9000/user/spark/ data/people.json") df.select("age").write.format("parquet").mode(SaveMode.Append).save("hdfs://master:9000/user/spark/data/ages.parquet")

7、直接使用sql查询数据源

val df = sqlContext.sql("SELECT * FROM parquet.`hdfs://master:9000/user/spark/data/ages.parquet`") df.map(x => "name:" + x(0)).foreach(println)

8、parquest文件的读写

val people = sc.textFile("hdfs://master:9000/user/spark/data/people.txt").toDF people.write.mode(SaveMode.Overwrite).parquet("hdfs://master:9000/user/spark/data/people.parquet") val parquetFile = sqlContext.read.parquet("hdfs://master:9000/user/spark/data/people.parquet") parquetFile.registerTempTable("parquetFile") val teenagers = sqlContext.sql("SELECT name FROM parquetFile") teenagers.map(t => "Name: " + t(0)).collect().foreach(println)

9、Schema Merging

val df1 = sc.makeRDD(1 to 5).map(i => (i, i * 2)).toDF("single", "double") df1.write.mode(SaveMode.Overwrite).parquet("hdfs://master:9000/user/spark/data/test_table/key=1") df2 = sc.makeRDD(6 to 10).map(i => (i, i * 3)).toDF("single", "triple") df2.write.mode(SaveMode.Overwrite).parquet("hdfs://master:9000/user/spark/data/test_table/key=2") df3 = sqlContext.read.option("mergeSchema", "true").parquet("hdfs://master:9000/user/spark/data/test_table") df3.printSchema() df3.show()

10、hive metastore

val sqlContext = new HiveContext(sc)sqlContext.setConf("spark.sql.shuffle.partitions","5") sqlContext.sql("use my_hive") sqlContext.sql("create table if not exists sogouInfo (time STRING,id STRING,webAddr STRING,downFlow INT,upFlow INT,url STRING) row format delimited fields terminated by '\t'") sqlContext.sql("LOAD DATA LOCAL INPATH '/root/testData/SogouQ1.txt' overwrite INTO TABLE sogouInfo") sqlContext.sql("select " +"count(distinct id) as c " +"from sogouInfo " +"group by time order by c desc limit 10").collect().foreach(println)

11、df from jdbc eg:mysql

val sqlContext = new SQLContext(sc) val jdbcDF = sqlContext.read.format("jdbc").options(Map("driver" -> "com.mysql.jdbc.Driver","url" -> "jdbc:mysql://192.168.0.65:3306/test?user=root&password=root","dbtable" -> "trade_total_info_copy")).load() jdbcDF.registerTempTable("trade_total_info_copy") sqlContext.sql("select * from trade_total_info_copy").foreach(println)

本文参与 腾讯云自媒体分享计划,分享自作者个人站点/博客。
原始发表:2017.03.31 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
相关产品与服务
大数据
全栈大数据产品,面向海量数据场景,帮助您 “智理无数,心中有数”!
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档