【 Spark编程基础】实验3

准备工作

启动Hadoop集群 & Spark

• 启动Hadoop集群start-all.sh
• 启动Sparkcd /usr/local/spark/spark-2.3.3-bin-hadoop2.6/# ./sbin/start-all.sh

实验数据说明。

• 数据为1970年到2016年，每年各球队的球员比赛数据统计，数据文件的格式如图1所示：
在这里插入图片描述

图1. NBA球员评估数据

• 篮球数据缩写说明如图2所示：
在这里插入图片描述
图2.篮球数据缩写说明

• 获取数据集：

数据集路径：https://staticfile.eec-cn.com/dataSet/systemLib/80c034eedda84db5aeec2a1558c51cc6.zip

• 将下载的数据集，解压缩到/home/data/目录下

mkdir /home/data
cd /home/data
wget https://staticfile.eec-cn.com/dataSet/systemLib/80c034eedda84db5aeec2a1558c51cc6.zip
unzip 80c034eedda84db5aeec2a1558c51cc6.zip

#1 集群启动

• 启动Hadoop集群

cd /opt/module/hadoop
./sbin/start-all.sh

• 启动Spark

 cd /opt/module/spark./sbin/start-all.sh

#2 上传实验数据到HDFS

• 创建数据目录

cd /opt/module/hadoop
hadoop fs -mkdir -p /home/student/data

• 上传数据集

hadoop fs -put /home/data/nba_data/* /home/student/data

#3 分析实现过程

启动IDEA，创建Python文件

启动IDEA

cd /opt/module/idea-IU-223.8836.41
./bin/idea.sh

登录自己的JetBrain账号
创建Python项目

实现过程

SparkSession

from pyspark import SparkConf
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
if __name__ == '__main__':conf = SparkConf().setAppName("spark sql demo").setMaster("local[*]")sparkSession = SparkSession.builder.config(conf=conf).getOrCreate()
ds = sparkSession.read.format('csv').load("hdfs://node1:8020/home/student/data/leagues_NBA_2016_per_game_per_game.csv")#查看表ds.show()#查看表结构ds.printSchema()
#查看某一列 类似于MySQL： select Player,Tm from peopleds.select("Player", "Tm").show()
#查看多列并作计算 #基于当前列进行加1ds.select("Player", ds.Age+1).show()
#设置过滤条件 ds.filter("Age > 21").show()
#做聚合操作 ds.groupBy("Tm").count().show()
#上述多个条件进行组合 select ta.age,count(*) from# (select name,age+1 as "age" from people) as ta where ta.age>21 group by ta.ageds.select("Player", (ds.Aage + 1).alias("Age")).filter("Age > 21").groupBy("Tm").count().show()#直接使用spark SQL进行查询#先注册为临时表ds.createOrReplaceTempView("nba")ds.cache()sqlDF = sparkSession.sql("SELECT * FROM nba")sqlDF.show()

SparkContext

from pyspark import SparkConf, SparkContext, SQLContext, Row
from pyspark.sql.types import IntegerType, StructType, StructField, StringType

if __name__ == '__main__':conf = SparkConf().setAppName("HiveDemo").setMaster("local")sc = SparkContext(conf=conf)# Spark1.0中访问SparkSQL的方式，现在还保留，为了向后兼容，Spark2.0以后用SparkSession可以代替sqlContext = SQLContext(sc)# 设置conf,配置AppName,运行的Master(这里设置为本地模式# 创建一个sc的SQLContext对象# 创建一个sqlcontext对象(也可以是SQLContext的子类对象,如 HiveContext)# 加载数据源datas = sc.textFile("hdfs://node1:8020/home/student/data/leagues_NBA_2016_per_game_per_game.csv")# RDD转换为DataFrame有两种方式:(这里使用了第二种)# 使用反射方式推断元数据# 使用编程接口来创建DataFrame.rowRDD = datas.map(lambda line: line.split(",")).map(lambda data: Row(int(data[0]), data[1], data[2]))# 创建出元素为ROW的RDD
# 流程简介:从原始的RDD创建一个元素为row的RDD;接下来创建一个structType,来代表ROW,最后将动态定义的# 元数据应用到RDD(ROW)上structType = StructType([StructField("Rk", IntegerType(), True),StructField("Player", StringType(), True),StructField("Age", IntegerType(), True),StructField("Tm", StringType(), True)# 通过编程的方式动态的构造元数据])# 通过sqlContext的createDataFrame方法,创建DataFrame,# 将row类型的RDD和数据结构structType结合到一起stuDF = sqlContext.createDataFrame(rowRDD, structType)stuDF.show()# show方法可以把里面的数据显示出来stuDF.registerTempTable("nba")# 注册为临时表,这样就可以使用SQL语句了.sqlContext.sql("select Player from nba where Tm='SAC'").show()