Spark SQL实战

简介: 一、程序 1 package sparklearning 2 3 import org.apache.log4j.Logger 4 import org.apache.spark.SparkConf 5 import org.

一、程序

 1 package sparklearning
 2 
 3 import org.apache.log4j.Logger
 4 import org.apache.spark.SparkConf
 5 import org.apache.spark.SparkContext
 6 import org.apache.spark.sql.SQLContext
 7 import org.apache.spark.storage.StorageLevel
 8 import org.apache.log4j.Level
 9 
10 object OnLineTradeStatistics {
11 
12   case class User(userID:String,gender:String,age:Int,registerDate:String,provice:String,career:String)
13   case class TradeDetail(tradeID:String, tradeDate:String,productID:Int,amount:Int,userID:String)
14   def main(args: Array[String]){
15       
16     //关闭不必要的日志显示
17       Logger.getLogger("org.apache.hadoop").setLevel(Level.ERROR)
18       Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
19       Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
20       
21       //设置应用程序
22       val conf=new SparkConf().setAppName("On Line Trade Data").setMaster("local")
23       val ctx=new SparkContext(conf)
24       val sqlCtx=new SQLContext(ctx)
25       import sqlCtx.implicits._
26       
27       //读文件  RDD-->DataFrame  
28       val userDF= ctx.textFile("/home/hadoop/data/on_line_trade_user.txt").map(_.split(" ")).map(u=>User(u(0),u(1),u(2).toInt,u(3),u(4),u(5))).toDF()
29       userDF.registerTempTable("user")
30       userDF.persist(StorageLevel.MEMORY_ONLY_SER)
31       
32       val tradeDF= ctx.textFile("/home/hadoop/data/on_line_trade_detail.txt").map(_.split(" ")).map(u=>TradeDetail(u(0),u(1),u(2).toInt,u(3).toInt,u(4))).toDF()
33       tradeDF.registerTempTable("trade")//生成临时表
34       tradeDF.persist(StorageLevel.MEMORY_ONLY_SER)
35       
36       val countOfTrade2016 = sqlCtx.sql("SELECT * FROM trade where tradeDate like '2016%'").count()
37       println("2016 total money: "+countOfTrade2016)
38   }
39 }

二、结果

当神已无能为力,那便是魔渡众生
liurio
+关注
目录
打赏
0
0
0
0
3
分享
相关文章
ClickHouse与大数据生态集成:Spark & Flink 实战
【10月更文挑战第26天】在当今这个数据爆炸的时代,能够高效地处理和分析海量数据成为了企业和组织提升竞争力的关键。作为一款高性能的列式数据库系统,ClickHouse 在大数据分析领域展现出了卓越的能力。然而,为了充分利用ClickHouse的优势,将其与现有的大数据处理框架(如Apache Spark和Apache Flink)进行集成变得尤为重要。本文将从我个人的角度出发,探讨如何通过这些技术的结合,实现对大规模数据的实时处理和分析。
237 2
ClickHouse与大数据生态集成:Spark & Flink 实战
Spark SQL向量化执行引擎框架Gluten-Velox在AArch64使能和优化
本文摘自 Arm China的工程师顾煜祺关于“在 Arm 平台上使用 Native 算子库加速 Spark”的分享,主要内容包括以下四个部分: 1.技术背景 2.算子库构成 3.算子操作优化 4.未来工作
|
3月前
|
SQL性能提升秘籍:5步优化法与10个实战案例
在数据库管理和应用开发中,SQL查询的性能优化至关重要。高效的SQL查询不仅可以提高应用的响应速度,还能降低服务器负载,提升用户体验。本文将分享SQL优化的五大步骤和十个实战案例,帮助构建高效、稳定的数据库应用。
231 3
|
3月前
|
SQL性能提升指南:五大优化策略与十个实战案例
在数据库性能优化的世界里,SQL优化是提升查询效率的关键。一个高效的SQL查询可以显著减少数据库的负载,提高应用响应速度,甚至影响整个系统的稳定性和扩展性。本文将介绍SQL优化的五大步骤,并结合十个实战案例,为你提供一份详尽的性能提升指南。
95 0
【赵渝强老师】Spark SQL的数据模型:DataFrame
本文介绍了在Spark SQL中创建DataFrame的三种方法。首先,通过定义case class来创建表结构,然后将CSV文件读入RDD并关联Schema生成DataFrame。其次,使用StructType定义表结构,同样将CSV文件读入RDD并转换为Row对象后创建DataFrame。最后,直接加载带有格式的数据文件(如JSON),通过读取文件内容直接创建DataFrame。每种方法都包含详细的代码示例和解释。
大数据-97 Spark 集群 SparkSQL 原理详细解析 Broadcast Shuffle SQL解析过程(一)
大数据-97 Spark 集群 SparkSQL 原理详细解析 Broadcast Shuffle SQL解析过程(一)
135 0
大数据-97 Spark 集群 SparkSQL 原理详细解析 Broadcast Shuffle SQL解析过程(二)
大数据-97 Spark 集群 SparkSQL 原理详细解析 Broadcast Shuffle SQL解析过程(二)
128 0
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
107 0
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")
|
7月前
|
SQL Server的并行实施如何优化?
【7月更文挑战第23天】SQL Server的并行实施如何优化?
185 13