大数据技术spark基础实验指导书

三盟Spark实验指导三盟Spark实验指导书三盟目录Scala部署及基本语法 5安装Scala 61.1进入安装目录 6Scala基础 72.1基础语法 7实验一三盟目录Scala部署及基本语法 5安装Scala 61.1进入安装目录 6Scala基础 72.1基础语法 7实验一结构 11实验二3.Scala数据结构 13Scala数据结构 3.4列表 14集合 14（Map） 15迭代器 17实验三4.Scala类和对象 20类 214.1简单的类 21对象 5.3单例对象 25伴生对象 25应用程序对象 26实验四6.Scala函数式编程WordCount 27实验操作 296.1创建需要统计的目录 29spark安装部署 31实验五1、检查环境hadoop集群环境 312、Spark部署 323、启动spark集群 43实验六SparkRDD编程 451.RDD编程 46实验七Spark单词统计实验 472、检查hadoop集群环境 492、创建项目 50三盟创建javamaven项目 51更改JDK版本 54配置pom.xml文件 58创建SparkWordCount类 64打包项目三盟创建javamaven项目 51更改JDK版本 54配置pom.xml文件 58创建SparkWordCount类 64打包项目 704、5、jar程序 765.1运行程序 76实验八SparkHbase读写实验 771、 检查hadoop集群环境 792、 Hbase创建studen表 802.1进入hbase环境 80创建......................................................................................................80表 812.4往表里数据 812.5使用scan命令查看表结构 823、创建项目 83创建javamaven项目 84更改JDK版本 87配置pom.xml文件 914、5、6、创建Scalajar包 98Spark运行打包jar文件 101java代码和配置文件（附） 104实验九Sparkstream实验 109实验准备 110创建Scalajar包 111创建Scala项目 111把scala项目Maven管理 1113Sparkjar文件 114java代码和配置文件（附） 116Stream类 116实验十SparkSQL实验 1191.实验准备 1205.创建Scalajar包 121三盟创建Scala项目 121把三盟创建Scala项目 121把scala项目Maven管理 1216Sparkjar文件 124java代码和配置文件（附） 126SQL类 126三盟实验一Scala【实验名称】Scala【实验目的】ScalaScala3.Scala三盟实验一Scala【实验名称】Scala【实验目的】ScalaScala3.Scala【实验要求】语句ScalaScala编写程序。【实验环境】1、1Centos6.82、实验所需安装包三盟【实验步骤】1.安装Scala1.1[root@localhost~]#cd/opt/hadoop-package/cpscala三盟【实验步骤】1.安装Scala1.1[root@localhost~]#cd/opt/hadoop-package/cpscala(scala-2.10.4.tgz)解压到/opt[root@localhosthadoop-package]#tar-zxvfscala-2.10.4.tgz-C/opt/scala[root@localhost]#cd/opt[root@localhostopt]#mvscala-2.10.4/scala设置环境变量，并是环境变量生效[root@localhost]#vi/etc/profile在文件最后加上下面环境配置使环境生效[root@localhost]#source/etc/profile修改配置截图：1.5scala测试scala[root@localhost]#scala安装如下图：三盟2.Scala值和变量Scalaval：是不可变的，在var：是可变的，时就必须被初始化，而且初始化以后就不能再赋值；的时候需要进行初始化，初始化以后还可以再次对其赋值。例如：不可变的字符串scala>valmyStr三盟2.Scala值和变量Scalaval：是不可变的，在var：是可变的，时就必须被初始化，而且初始化以后就不能再赋值；的时候需要进行初始化，初始化以后还可以再次对其赋值。例如：不可变的字符串scala>valmyStr="HelloWorld"scala>valmyStr2:String="HelloWorld"scala>valmyStr3:java.lang.String="HelloWorld"例如：可变的复数scala>varmyPrice:Double=9.9scala>myPrice =9.9三盟2.1.2ScalaChar、Short、Int、Long、Float、DoubleBooleanaaaasca三盟2.1.2ScalaChar、Short、Int、Long、Float、DoubleBooleanaaaascaaIntscala.Int。对于字符串，Scalajava.lang.String操作符：在Scala中，可以使用加(+)、减(-)、乘(*)、除(/)、余数（%）等操作符，而且，这些操作符就是。例如，53(5).+(3)是等价的，也就是说：ab 等价于a.bscala>varsum1=5+3等价于scala>varsum1=(5).+(3)三盟2.1.3Rangefor，i15，这时就可以采RangeRangeBigIntBigDecimal（1）155，1命令如下scala>1to5（2）15三盟2.1.3Rangefor，i15，这时就可以采RangeRangeBigIntBigDecimal（1）155，1命令如下scala>1to5（2）155，1scala>1until5（3）11010，2scala>1to10by2（4）Float0.5f5.9f0.3fscala>0.5fto5.9fby0.3f2.1.4台输入输出语句为了从台读写数据，可以使用以为前缀的Long、，Char：Int、Boolean、Line，Byte、Short、Float、98以带一个字符串参数的提示没有参数，Line三盟scala.io.StdIn用为了向台输出，常用的两个函数是print()和println()，可以直接输出字符串或者其它数据类型scala>println("231")ScalaCprintf()函数2.1.5读写文件1.写入文件Scalajava.io.PrintWriter三盟scala.io.StdIn用为了向台输出，常用的两个函数是print()和println()，可以直接输出字符串或者其它数据类型scala>println("231")ScalaCprintf()函数2.1.5读写文件1.写入文件Scalajava.io.PrintWriterscala>importjava.io.PrintWriterscala>valout=newPrintWriter("output.txt")scala>for(i<-1to5)out.println(i)scala>out. ()如果我们想把文件保存到一个指定的目录下，就需要给出文件路径scala>importjava.io.PrintWriter三盟scala>valout=newPrintWriter("/root/output.txt")scala>for(i<-1to三盟scala>valout=newPrintWriter("/root/output.txt")scala>for(i<-1to5)out.println(i)scala>out. ()2.文件Scala.io.SourcegetLines实现对文件中所有行的scala>importscala.io.Sourcescala>varinputFile=Source.fromFile("output.txt")scala>vallines=inputFile.getLinesfor(line<-lines)println(line)2.2结构2.2.1ifscala>valx=6scala>if(x>0) 三盟 println("false") }2.2.2whilescala>三盟 println("false") }2.2.2whilescala>valx=6scala>while(x>0){x-=1printf("xis%d",x) }2.2.3forScalaforscala>for(i<-1to5)println(i)三盟实验二Scala【实验名称】Scala【实验目的】Scala熟悉掌握列表、集合、、迭代器、数组、元组等常见的数据类型。【实验要求】在实验结束时能熟悉掌握列表、集合、、迭代器、数组、元组等常见数据操作三盟实验二Scala【实验名称】Scala【实验目的】Scala熟悉掌握列表、集合、、迭代器、数组、元组等常见的数据类型。【实验要求】在实验结束时能熟悉掌握列表、集合、、迭代器、数组、元组等常见数据操作【实验环境】1scalaCentos6.8三盟【实验步骤】3.Scala3.1列表列表是一种共享相同类型的不可变的对象序列。既然是一个不可变的集合，Scala的List不同于Java的java.util.List，scala的List一旦被定义，其值就不能改变，因此时必须初始化scala>varstrList=List("BigData","Hadoop","Spark")Listheadtailhead来获取tail的新列表，这体现出列表具有递归的链表结构strList.head三盟【实验步骤】3.Scala3.1列表列表是一种共享相同类型的不可变的对象序列。既然是一个不可变的集合，Scala的List不同于Java的java.util.List，scala的List一旦被定义，其值就不能改变，因此时必须初始化scala>varstrList=List("BigData","Hadoop","Spark")Listheadtailhead来获取tail的新列表，这体现出列表具有递归的链表结构strList.headigDat，strList.ailtadoop""Sark")3.2集合集合(st)是不重复元素的容器（n。列表中的元素是按照的先后顺序来组织元素的顺序，而是以“”对元素的值进行组织，所以，它你快速地找到某个元素集合可变集和不可变集，分别位于变集3.2.1不可变集scala>varmySet=Set("Hadoop","Spark")scala>mySet+="Scala"三盟3.2.2可变集scala>importscala.collection.mutable.Setscala>valmyMutableSet=Set("Database","BigData")scala>myMutableSet+="CloudComputing"3.3（Map）以根据键来对值进行快速的检索采用了类继承机制提供了可变的和不可变的两种版本的在包scala.collection.mutable和scala.collection.immutable里。默认情况下，Scala中使用不可变的。如果想使用可变，必须明确地导入scala.collection.mutable.Map3.3.1不可变Map三盟3.2.2可变集scala>importscala.collection.mutable.Setscala>valmyMutableSet=Set("Database","BigData")scala>myMutableSet+="CloudComputing"3.3（Map）以根据键来对值进行快速的检索采用了类继承机制提供了可变的和不可变的两种版本的在包scala.collection.mutable和scala.collection.immutable里。默认情况下，Scala中使用不可变的。如果想使用可变，必须明确地导入scala.collection.mutable.Map3.3.1不可变Mapscala>valuniversity=Map("XMU"->"XiamenUniversity","THU"->"TsinghuaUniversity","PKU"->"PekingUniversity")如果要获取中的值，可以通过键来获取scala>println(university("XMU"))对于这种contains确定键是否scala>valxmu=if(university.contains("XMU"))university("XMU")else0scala>println(xmu)三盟不可变，是无法更新中的元素的，也无法增加新的元素。如果要更新的元素，就需要定义一个可变的3.3.2可变Mapscala>importscala.collection.mutable.Mapscala>valuniversity2=三盟不可变，是无法更新中的元素的，也无法增加新的元素。如果要更新的元素，就需要定义一个可变的3.3.2可变Mapscala>importscala.collection.mutable.Mapscala>valuniversity2=Map("XMU"->"Xia niversity","THU"->"TsinghuaUniversity","PKU"->"PekingUniversity")scala>println(university2("XMU"))也可以使用+=操作来添加新的元素scala>university2+=("TJU"->"TianJingUniversity")3.3.3循环遍历scala>for((k,v)<-university2)printf(Codeis:%sandnameis:%s\n,k,v)三盟或者，也可以只遍历kvscala>for(k<-university2.keys)println(k)scala>for(v<-university2.values)三盟或者，也可以只遍历kvscala>for(k<-university2.keys)println(k)scala>for(v<-university2.values)println(v)3.4迭代器Scala（Iterator）不是一个集合，但是，提供了集合的法用于检测是否还有下一个元素scala>valiter=Iterator("Hadoop","Spark","Scala")scala>while(iter.hasNext){println(iter.next())}1.5数组（Array）的数据结构。ScalaArray[T]TScala类型，可以通过显式指定类型或者通过隐式推断来实例化一个数组。1.5.1普通数组三盟scala>varintValueArr=newArray[Int](3)scala>三盟scala>varintValueArr=newArray[Int](3)scala>intValueArr(0)=12scala>intValueArr(1)=45scala>intValueArr(2)=331.5.2函数ofDimArrayofDimscala>val myMatrix=Array.ofDim[Int](3,4)//类型实际就是Array[Array[Int]]scala> val myCube = Array.ofDim[String](3,2,4) //类型实际Array[Array[Array[Int]]]是可以使用多级圆括号来素数组的元素，例如myMatrix(0)(1)返回第一行第二列的元1.5.3定长数组采用Array类型定义的数组属于定长数组，其数组长度在初始化后就不能改变。如果要定例如下：scala>import scala>valaMutableArr=ArrayBuffer(10,20,30)scala>aMutableArr+=40三盟scala>aMutableArr.insert(2,60,40)三盟scala>aMutableArr.insert(2,60,40)scala>aMutableArr-=40temp=aMutableArr.remove(2)1.6元组（Tuple）以包含不同类型的元素scala>valtuple=("hello",2015,44.7)scala>println(tuple._1)scala>println(tuple._2)scala>println(tuple._3)三盟实验三Scala【实验名称】Scala【实验目的】1.2.3.三盟实验三Scala【实验名称】Scala【实验目的】1.2.3.ScalaScala了解单例对象、伴生对象、应用程序对象的特征和区别。【实验要求】实验结束后学生能够熟悉使用Scala文件的编译和执行，能够熟悉使用类的属性和方法，了解单例对象、伴生对象、应用程序对象的特征和区别【实验环境】1scalaCentos6.8三盟【实验步骤】4.类4.1简单的类最简单的类的定义形式是：new给类增加字段和classCounter{private三盟【实验步骤】4.类4.1简单的类最简单的类的定义形式是：new给类增加字段和classCounter{privatevarvalue=0defincrement():Unit={value+=1}defcurrent():Int={value}}创建对象classCounter{privatevarvalue=0defincrement():Unit={value+=1}defcurrent():Int={value}}下面我们新建对象，并调用其中的：valmyCounter=newCountermyCounter.increment()三盟Scala1.4编译和执行时，是可以省略名后面的圆括号的1.4.1编辑文件内容viTestCounter.scala内容如下class三盟Scala1.4编译和执行时，是可以省略名后面的圆括号的1.4.1编辑文件内容viTestCounter.scala内容如下classCounter{privatevarvalue=0defincrement():Unit={value+=1}defcurrent():Int={value}}valmyCounter=newCountermyCounter.increment()1.4.2在LinuxShellscala1.4.3输出结果上面命令执行后，会在屏幕输出11.4.4JVMJVMscalac1.编辑文件内容[root@hadoop1]#viTestCounterJVM.scala内容如下classCounter{privatevarvalue=0defincrement():Unit={value+=1}defcurrent():Int={value}三盟}objectMyCounter{defmain(args:Array[String]){valmyCounter=newCounterMyCounter.increment()println(myCounter.current)}}2.scalac三盟}objectMyCounter{defmain(args:Array[String]){valmyCounter=newCounterMyCounter.increment()println(myCounter.current)}}2.scalacscala[root@hadoop1]#scalacTestCounterJVM.scala[root@hadoop1]#scala-classpath.MyCounter1.5gettersetter给类中的字段设置值以及Javagettersetter实现的Scalagettersetter的实现，但是并没有定义成get 和setvalue变成私有字段以后，Scala又没有提供getter和setter，怎么可以valueScalagettersettervaluevalue_=，具体如下：，分别叫做1.5.1编辑文件内容viTestCounterJVM.scala内容如下objectMyCounter{defmain(args:Array[String]){ valmyCounter=newCounter() println(myCounter.value) myCounter.value=3三盟}}classCounter{privatevarprivateValue=三盟}}classCounter{privatevarprivateValue=0defvalue=privateValuedefvalue_=(newValue:Int){ if(newValue>0) =newValue}defincrement(temp:Int):Unit={value+=temp}defcurrent():Int={value}}1.5.2删除之前生成的class[root@hadoop1]#rm-rfMyCounter*1.5.3使用scalac[root@hadoop1]#scalacTestCounterJVM.scala[root@hadoop1]#scala-classpath.MyCounter04。5.对象三盟5.1单例对象ScalaJavaJava同样的功能。下面是单例对象的定义：privatevarlastId=0 defnew Id()={ lastId+=1 lastId }}println("1persionidis", .new Id())println("2persion三盟5.1单例对象ScalaJavaJava同样的功能。下面是单例对象的定义：privatevarlastId=0 defnew Id()={ lastId+=1 lastId }}println("1persionidis", .new Id())println("2persionidis", .new println("3persionidis", .new 5.2伴生对象Java生对象来实现。和静态Scala类和它的伴生对象必须一个文件中，而且可以相互私有成员（字段和。2.2.1test.scala在该文件中输入如下代码：class {privatevalid=.newId() privatevarname=""object{三盟 defthis(name:String){ this() =name } definfo(){printf("Theidof三盟 defthis(name:String){ this() =name } definfo(){printf("Theidof%sis%d.\n",name,id)}}object { privatevarlastId=0 //一个人的 编号 privatedefnew Id()={ lastId+=1 lastId } defmain(args:Array[String]){ val 1=new ("Ziyu") val 2=new ("Minxing") 1.info() 2.info() }}2.2.2编译执行[root@hadoop1]#scalactest.scala[root@hadoop1]#scalaPersion5.3应用程序对象Scalamaintest.scala5.3.1test.scala[root@hadoop1]#rm-rftest.scala三盟[root@hadoop1]#vitest.scala文件中输入如下代码：objectHelloWorld{ defmain(args:Array[String]){三盟[root@hadoop1]#vitest.scala文件中输入如下代码：objectHelloWorld{ defmain(args:Array[String]){ println("Hello,World!") }}为了运行上述代码，我们现在可以使用两种不同的。第法:直接使用scala命令scala5.3.2scala[root@hadoop1]#scalatest.scala5.3.3[root@hadoop1]#scalactest.scala[root@hadoop1]#scala-classpath.HelloWorld实验四ScalaWordCount【实验名称】WordCount三盟【实验目的】对前面知识的消化理解。【实验要求】Scala三盟【实验目的】对前面知识的消化理解。【实验要求】Scala【实验环境】1scalaCentos6.8三盟【实验步骤】6.实验操作6.1创建需要统计的目录[root@hadoop1]#mkdirwordcount三盟【实验步骤】6.实验操作6.1创建需要统计的目录[root@hadoop1]#mkdirwordcount1.2在目录下分别创建word1.txtword2.txt[root@hadoop1]#viwordcount/word1.txt[root@hadoop1]#viwordcount/word2.txt你可以在文本文件中随意输入一些单词，用空格隔开1.3新建一个testWordCount.scala[root@hadoop1]#vitestWordCount.scala内容如下importjava.io.Fileimportscala.io.SourceobjectWordCount{defmain(args:Array[String]):Unit={ valdirfile=newFile("/root/wordcount") valfiles=dirfile.listFiles三盟 for(file<-files)println(file) val三盟 for(file<-files)println(file) vallistFiles=files.toList "). foreach( word=>{ if(wordsMap.contains(word)){ wordsMap(word)+=1 }else{ wordsMap+=(word->1) } } ) ) ) println(wordsMap) println(key+":"+value)}}1.4执行Scala[root@hadoop1]#scalatestWordCount.scala执行截图：listFiles.foreach( 三盟实验五spark【实验名称】spark【实验目的】Spark的SparkSparkSparkSpark三盟实验五spark【实验名称】spark【实验目的】Spark的SparkSparkSparkSparkSparkSparkHDFS【实验要求】Master，hadoop2、hadoop3WorkerSpark进试。【实验环境】【实验步骤】1、检查环境hadoop集群环境3hadoopjpshadoop3.1注意：如果hadoop没有安装完成请安装第1课进行安装或执行自动部署脚本三盟hadoop1jpshadoop2jpshadoop3三盟hadoop1jpshadoop2jpshadoop3jps2、Spark部署hadoop1三盟2.1安装包的目录在终端执行命令：cd/opt/hadoop-package2.2spark三盟2.1安装包的目录在终端执行命令：cd/opt/hadoop-package2.2spark/opt在终端执行命令：tarzxvfspark-2.3.1-bin-hadoop2.7.gz-C/opt/解压完成的截图：三盟2.3scala/opt三盟2.3scala/opt在终端执行命令：tarzxvfscala-2.11.8.tgz-C/opt/解压完成的截图：三盟2.4opt在终端执行命令：cd/opt2.5三盟2.4opt在终端执行命令：cd/opt2.5spark在终端执行命令：mvspark-2.3.1-bin-hadoop2.7spark-scala在终端执行命令：mvscala-2.11.8scala三盟2.7spark2.8spark-env.sh.templatespark-env.sh三盟2.7spark2.8spark-env.sh.templatespark-env.sh2.9spark-env.shexportSCALA_HOME=/opt/scalaexportHADOOP_HOME=/opt/hadoop-3.1.0exportHADOOP_CONF_DIR=/opt/hadoop-3.1.0/etc/hadoopexportSPARK_MASTER_IP=hadoop1exportSPARK_WORKER_CORES=2SPARK_WORKER_INSTANCES=1exportclasspath)三盟加入代码后的截图：2.10slaves.templateslaves在终端执行命令：cpslaves.templateslaves2.11三盟加入代码后的截图：2.10slaves.templateslaves在终端执行命令：cpslaves.templateslaves2.11slaves在终端执行命令：vimslaves三盟localhost三盟localhostlocalhost三盟删除完成后再该文件的最后面加入下面两行代码：hadoop2hadoop3加入代码后的截图：2.12同步三盟删除完成后再该文件的最后面加入下面两行代码：hadoop2hadoop3加入代码后的截图：2.12同步spark、scala文件到hadoop2hadoop3scalahadoop2在终端执行命令：scp-ropt/scalaroot@hadoop2:/opt同步完成的截图：三盟scalahadoop3在终端执行命令：scp三盟scalahadoop3在终端执行命令：scp-ropt/scalaroot@hadoop3:/opt同步完成的截图：三盟sparkhadoop2在终端执行命令：scp三盟sparkhadoop2在终端执行命令：scp-ropt/spark-2.3.1/root@hadoop2:/opt同步完成的截图：三盟sparkhadoop3在终端执行命令：scp三盟sparkhadoop3在终端执行命令：scp-ropt/spark-2.3.1/root@hadoop3:/opt同步完成的截图：三盟3spark集群3.1启动三盟3spark集群3.1启动sparkhadoop1三盟启动截图：三盟启动截图：三盟实验六SparkRDD【实验名称】Spark三盟实验六SparkRDD【实验名称】SparkRDD【实验目的】RDDRDDRDDScala【实验要求】max、first、distinct、foreachapi【实验环境】Centos6.8Jdk1.8+scalaspark三盟【实验步骤】1.RDD1.1RDD1.1.1进入到sparkbin三盟【实验步骤】1.RDD1.1RDD1.1.1进入到sparkbin[root@hadoop1]#cd/opt/spark-2.3.1/bin/1.1.2切换回spark-shell[root@hadoop1bin]#spark-shell1.1.3通过scRDD本地：scala>vallines=sc.textFile("file:///root/wordcount.txt")Hdfs:scala>vallines=sc.textFile("hdfs://hadoop1:9000/user/root/wordcount.txt")scala>vallines=sc.textFile("/user/root/wordcount.txt")scala>vallines=sc.textFile("wordcount.txt")hdfs三盟1.2wordcount文字统计scala>vallines=sc.textFile("file:///root/wordcount.txt")scala>lines.flatMap(_.split("")).map((_,完成如图：1.3RDD其它操作scala>三盟1.2wordcount文字统计scala>vallines=sc.textFile("file:///root/wordcount.txt")scala>lines.flatMap(_.split("")).map((_,完成如图：1.3RDD其它操作scala>lines.count()实验七Spark单词统计实验【实验名称】scala>lines.first()三盟Spark【实验目的】Scala三盟Spark【实验目的】ScalaSparkSparkWordCount【实验要求】ScalaSparkSparkWordCount,然后能够spark-shell【实验环境】三盟【实验步骤】2hadoop集群环境3hadoop三盟【实验步骤】2hadoop集群环境3hadoopjpshadoop注意hadoop1op3.13.1注意hadoop1hadoop1jpshadoop2jpshadoop3jps三盟2、创建项目hadoop1eclipse三盟2、创建项目hadoop1eclipse三盟2.1javamaven项目三盟2.1javamaven项目FileNewOther三盟在弹出的页面中按下图步骤操作三盟在弹出的页面中按下图步骤操作三盟三盟三盟2.2JDK版本三盟2.2JDK版本Properties三盟三盟三盟三盟三盟三盟三盟2.3pom.xml文件在创建pom.xml三盟2.3pom.xml文件在创建pom.xmlpom.xml三盟在打开的pom.xml文件中在28行附近找到</dependency>，然后在这一行的最前面回车换行三盟在打开的pom.xml文件中在28行附近找到</dependency>，然后在这一行的最前面回车换行换好行的截图：三盟打开桌面的【b8.3三盟打开桌面的【b8.3Spark打开【8.3Spark.docx】文件三盟三盟pom.xml的代码三盟eclipsepom.xml三盟eclipsepom.xml三盟黏贴截图：三盟黏贴截图：三盟保存2.4SparkWordCount三盟保存2.4SparkWordCount类SparkWordCount-testsrc/main/java右击创建类三盟三盟三盟删除里面的内容三盟删除里面的内容三盟删除后的截图：打开桌面的【b8.3Spark三盟删除后的截图：打开桌面的【b8.3Spark打开【8.3Spark.docx】文件三盟三盟束中间的代码三盟三盟三盟保存4、打包项目三盟保存4、打包项目Export三盟JARfile三盟JARfileNext三盟Next三盟Next三盟三盟三盟class,OK三盟class,OK然后Finish三盟OKOK三盟OKOK三盟5、运行打包jar程序5.1运行程序hadoo1sparkbin三盟5、运行打包jar程序5.1运行程序hadoo1sparkbin运行程序在终端执行命令：spark-submit--classfile:///root/wordcount.txt执行如图：三盟至此，本实验结束实验八SparkHbase读写实验【实验名称】SparkHbase【实验目的】三盟至此，本实验结束实验八SparkHbase读写实验【实验名称】SparkHbase【实验目的】SparkSparkHbase【实验要求】要求实验结束时，每个学生能够正确运行本实验所写的jar包程序，并能够熟悉掌握SparkHbase【实验环境】1hadoop2hbase三盟3spark三盟3spark三盟【实验步骤】1、hadoop集群环境3三盟【实验步骤】1、hadoop集群环境3hadoopjpshadoop注意hadoop1op3.13.1注意hadoop1hadoop1jpshadoop2jpshadoop3jps三盟、表hadoop12.1三盟、表hadoop12.1hbase环境在终端执行命令：hbaseshell2.2查看表是否hbase(main):007:0>list三盟2.3创建表hbase(main):007:0>create'hbase(main):007:0>describe'','info''2.4往表里数据//首先录入表的第一个学生hbase(main):007:0>三盟2.3创建表hbase(main):007:0>create'hbase(main):007:0>describe'','info''2.4往表里数据//首先录入表的第一个学生hbase(main):007:0>put'hbase(main):007:0>put'hbase(main):007:0>put'','1','info:name','Xueqian'','1','info:gender','F'','1','info:age','23'三盟//然后录入hbase(main):007:0>put'hbase(main):007:0>put'hbase(main):007:0>put三盟//然后录入hbase(main):007:0>put'hbase(main):007:0>put'hbase(main):007:0>put'','2','info:gender','M'','2','info:age','24'2.5scan命令查看表结构hbase(main):007:0>scan'这样我们就往表里增加了两条'三盟3、创建项目hadoop1eclipse三盟3、创建项目hadoop1eclipse三盟3.1javamaven项目三盟3.1javamaven项目FileNewOther三盟在弹出的页面中按下图步骤操作三盟在弹出的页面中按下图步骤操作三盟三盟三盟3.2JDK版本三盟3.2JDK版本Properties三盟三盟三盟三盟三盟三盟三盟3.3pom.xml文件在创建pom.xml三盟3.3pom.xml文件在创建pom.xmlpom.xml三盟在打开的pom.xml文件中在28行附近找到</dependency>，然后在这一行的最前面回车换行三盟在打开的pom.xml文件中在28行附近找到</dependency>，然后在这一行的最前面回车换行换好行的截图：三盟打开桌面的【b8.3三盟打开桌面的【b8.3Spark打开【8.3Spark.docx】文件三盟三盟pom.xml的代码三盟eclipsepom.xml三盟eclipsepom.xml三盟黏贴截图：三盟黏贴截图：三盟保存三盟保存三盟4ScalajarScala三盟4ScalajarScala项目file >>New >>ScalaProject把scala项目Maven管理右键项目>>Configure >>ConverttoMavenProject2.3配置pom.xml文件加载jar包三盟2.4创建ScalaObject实例2.5创建三盟2.4创建ScalaObject实例2.5创建Hbase2.6设置表名和创建APIHadoopRDD连接三盟2.7循环表的字段2.8修改JavaBuildPathBuildPathIncluded改为**/*.*三盟2.7循环表的字段2.8修改JavaBuildPathBuildPathIncluded改为**/*.*main语言否则编java语言编译2.9Maven打包右键项目 >>>Runas>>>Mavenclean右键项目 >>>Runas>>>Maveninstall下如图target下出现jar包三盟5、Spark运行打包jar3.1将打包jar上传到linux3.1.1linuxConnect三盟5、Spark运行打包jar3.1将打包jar上传到linux3.1.1linuxConnectSFTPSessionsftp3.1.2jarsftp三盟3.1.3然后进入到sparkbin[root@hadoop1]#cd/opt/spark-2.3.1/bin/3.1.4使用spark-submit三盟3.1.3然后进入到sparkbin[root@hadoop1]#cd/opt/spark-2.3.1/bin/3.1.4使用spark-submitspark：[root@hadoop1 bin]# spark-submit --master local --classwordcount.SparkOperateHBase/root/WordCountScala-0.0.1-SNAPSHOT.jar写入：[root@hadoop1 bin]# spark-submit --master local --classwordcount.SparkWriteHBase/root/WordCountScala-0.0.1-SNAPSHOT.jar--class是你的类的具体地址默认是hdfsmaster默认是本地如下图三盟写入如下图：查看可以通过8088端口去看到刚刚的操作三盟写入如下图：查看可以通过8088端口去看到刚刚的操作三盟至此，本实验结束6、java代码和配置文件（附）Hbase类4.1packagewordcountimportorg.apache.hadoop.conf.Configurationimportorg.apache.hadoop.hbase._importorg.apache.hadoop.hbase.._importorg.apache.hadoop.hbase.mapreduce.TableInputFormatimportorg.apache.hadoop.hbase.util.Bytesimportorg.apache.spark.SparkContextimportorg.apache.spark.SparkContext._importorg.apache.spark.SparkConfobjectSparkOperateHBase三盟至此，本实验结束6、java代码和配置文件（附）Hbase类4.1packagewordcountimportorg.apache.hadoop.conf.Configurationimportorg.apache.hadoop.hbase._importorg.apache.hadoop.hbase.._importorg.apache.hadoop.hbase.mapreduce.TableInputFormatimportorg.apache.hadoop.hbase.util.Bytesimportorg.apache.spark.SparkContextimportorg.apache.spark.SparkContext._importorg.apache.spark.SparkConfobjectSparkOperateHBase{defmain(args:Array[String]){三盟valconf=HBaseConfiguration.create()/*conf.set("hbase.zookeeper.quorum","hadoop1")conf.set("perty.Port","2181")*/valsc=newSparkContext(newSparkConf())//设置的表名conf.set(TableInputFormat.INPUT_TABLE,"1")valstuRDD=sc.newAPIHadoopRDD(conf,classOf[TableInputFormat],classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],classOf[org.apache.hadoop.hbase..Result])stuRDD.cache()val三盟valconf=HBaseConfiguration.create()/*conf.set("hbase.zookeeper.quorum","hadoop1")conf.set("perty.Port","2181")*/valsc=newSparkContext(newSparkConf())//设置的表名conf.set(TableInputFormat.INPUT_TABLE,"1")valstuRDD=sc.newAPIHadoopRDD(conf,classOf[TableInputFormat],classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],classOf[org.apache.hadoop.hbase..Result])stuRDD.cache()valcount=stuRDD.count()println("sRDDCount:"+count)//遍历输出stuRDD.foreach({case(_,result)=>valkey=Bytes.toString(result.getRow)valname=Bytes.toString(result.getValue("info".getBytes,"name".getBytes))valgender=Bytes.toString(result.getValue("info".getBytes,"gender".getBytes))valage=Bytes.toString(result.getValue("info".getBytes,"age".getBytes))println("***")println("Rowkey:"+key+"Name:"+name+"Gender:"+gender+"Age:"+age)println("***")})三盟}}4.2写入Hbase类packagewordcountimportorg.apache.hadoop.hbase.HBaseConfigurationimportorg.apache.hadoop.hbase.mapreduce.TableOutputFormatimportorg.apache.spark._importorg.apache.hadoop.mapreduce.Jobimportorg.apache.hadoop.hbase.io.ImmutableBytesWritableimportorg.apache.hadoop.hbase..Resultimportorg.apache.hadoop.hbase..Putimportorg.apache.hadoop.hbase.util.BytesobjectSparkWriteHBase{defmain(args:Array[String]):Unit={valsparkConf=newSparkConf().setAppName("SparkWriteHBase").setMaster("local")valsc三盟}}4.2写入Hbase类packagewordcountimportorg.apache.hadoop.hbase.HBaseConfigurationimportorg.apache.hadoop.hbase.mapreduce.TableOutputFormatimportorg.apache.spark._importorg.apache.hadoop.mapreduce.Jobimportorg.apache.hadoop.hbase.io.ImmutableBytesWritableimportorg.apache.hadoop.hbase..Resultimportorg.apache.hadoop.hbase..Putimportorg.apache.hadoop.hbase.util.BytesobjectSparkWriteHBase{defmain(args:Array[String]):Unit={valsparkConf=newSparkConf().setAppName("SparkWriteHBase").setMaster("local")valsc=newSparkContext(sparkConf)valtablename=""sc.hadoopConfiguration.set(TableOutputFormat.OUTPUT_TABLE,tablename)valjob=newJob(sc.hadoopConfiguration)job.setOutputKeyClass(classOf[ImmutableBytesWritable])job.setOutputValueClass(classOf[Result])job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])=sc.makeRDD(Array("3,Rongcheng,M,26","4,Guanhua,M,27"))//构建两行valrdd=indataRDD.map(.split(',')).map{arr=>{valputnewPut(Bytes.toBytes(arr(0//行健的值三盟put.add(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes(arr(1)))列的值put.add(Bytes.toBytes("info"),Bytes.toBytes("gender"),Bytes.toBytes(arr(2)))列的值put.add三盟put.add(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes(arr(1)))列的值put.add(Bytes.toBytes("info"),Bytes.toBytes("gender"),Bytes.toBytes(arr(2)))列的值put.add(Bytes.toBytes("info"),Bytes.toBytes("age"),Bytes.toBytes(arr(3)))//info:age列的值(newImmutableBytesWritable,put)}}rdd.saveAsNewAPIHadoopDataset(job.getConfiguration())}}4.3pom.xml配置文件<properties><jdk.version>1.8</jdk.version><logback.version>1.1.2</logback.version><slf4j.version>1.7.7</slf4j.version><junit.version>4.11</junit.version><spark.version>2.1.0</spark.version><hadoop.version>2.6.5</hadoop.version><hbase.version>1.2.6</hbase.version></properties><dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.10</artifactId><version>2.1.0</version></dependency>三盟<dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-</artifactId><version>2.6.5</version></dependency><dependency><groupId>org.apache.hadoop</三盟<dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-</artifactId><version>2.6.5</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>2.6.5</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>2.6.5</version></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-</artifactId><version>1.2.6</version></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-server</artifactId><version>1.2.6</version></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase</artifactId><version>1.2.6</version><type>pom</type>三盟</dependency></dependencies>实验九Sparkstream【实验名称】Spark三盟</dependency></dependencies>实验九Sparkstream【实验名称】Sparkstream【实验目的】区别版本的WordCountMapReduce版本的WordCount【实验要求】词数目。【实验环境】Centos6.8Jdk1.8+安装好spark三盟【实验步骤】1.实验准备1.1HDFS三盟【实验步骤】1.实验准备1.1HDFS命令如下：[root@hadoop1]#hdfsdfs-mkdir/input三盟2.创建Scalajar2.1 创建Scalafile>>New>>Scala三盟2.创建Scalajar2.1 创建Scalafile>>New>>ScalaProject2.2把scalaMaven右键项目>>Configure >>ConverttoMavenProject2.3配置pom.xmljar三盟2.4ScalaObject2.5Stream2.6流文件计算三盟2.4ScalaObject2.5Stream2.6流文件计算三盟2.7启动stream2.8修改JavaBuildPathBuildPath中的Included三盟2.7启动stream2.8修改JavaBuildPathBuildPath中的Included改为**/*.*main的时候出现Runas出现scalaApplication，则表示scalajava2.9Maven打包右键项目 >>>Runas>>>Mavenclean右键项目 >>>Runas>>>Maveninstall下如图targetjar三盟3.Sparkjar3.1将打包jarlinux3.1.1linuxConnect三盟3.Sparkjar3.1将打包jarlinux3.1.1linuxConnectSFTPSessionsftp3.1.2jarsftp三盟3.1.3然后进入到sparkbin[root@hadoop1]#cd/opt/spark-2.3.1/bin/3.1.4使用spark-submit三盟3.1.3然后进入到sparkbin[root@hadoop1]#cd/opt/spark-2.3.1/bin/3.1.4使用spark-submitspark[root@hadoop1 bin]# spark-submit --master local --class/input--class是你的类的具体地址默认是hdfsmaster3.1.5hdfsinput[root@hadoop1]#hdfsdfs-put/root/wordcount.txt/input[root@hadoop1]#hdfsdfs-put/root/people.txt/input如下图三盟2020文件流至此，本实验结束4.java（附）4.1 Streampackagewordcountimportorg.apache.spark.SparkConfimport三盟2020文件流至此，本实验结束4.java（附）4.1 Streampackagewordcountimportorg.apache.spark.SparkConfimportorg.apache.spark.streaimportorg.apache.spark.strea.Strea.SecondsContextimportorg.apache.spark.storage.StorageLevelobjectHdfsWordCount{defmain(args:Array[String]){if(args.length<1){HdfsWordCount<hdfsdir>")System.exit(1)}三盟//Createthecontextwitha5second三盟//Createthecontextwitha5secondbatchsizevalsparkConf=newSparkConf().setAppName("HdfsWordCount")valssc=newStreaSeconds(20))vallines=ssc.textFileStream(args(0))valwordCounts=lines.flatMap(_.split("")).map((_,1)).reduceByKey(_+_)wordCounts.print()ssc.start()ssc.awaitTermination()}}4.2pom.xml<properties><jdk.version>1.8</jdk.version><logback.version>1.1.2</logback.version><slf4j.version>1.7.7</slf4j.version><junit.version>4.11</junit.version><spark.version>2.1.0</spark.version><hadoop.version>2.6.5</hadoop.version><hbase.version>1.2.6</hbase.version></properties><dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.10</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-</artifactId><version>${hadoop.version}</version></dependency>三盟<dependency><groupId>org.apache.hadoop</groupId三盟<dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-</artifactId><version>${hbase.version}</version></dependency><dependency><artifactId>hbase-server</artifactId><version>${hbase.version}</version></dependency><dependency><artifactId>hbase</artifactId><version>${hbase.version}</version><type>pom</type></dependency></dependencies>三盟实验十SparkSQL【实验名称】Spark三盟实验十SparkSQL【实验名称】SparkSQL【实验目的】SparkSQLSparkSQLSQL【实验要求】在实验结束后建立数据库，创建数据表的数据结构SparkSQLSQLSparkSQL【实验环境】Centos6.8Jdk1.8+安装好spark三盟【实验步骤】1.实验准备1.1创建people.tx