spark快速
spark通用
spark开放
2009 RAD实验室,引入内存存储 2010 开源 2011 AMP实验室,Spark Streaming 2013 Apache顶级项目
Spark Core:
Spark SQL:
Spark Streaming:
Spark Mlib:
GraphX:
Spark Cluster Managers:
生态系统、各司其职 Spark需要借助HDFS进行持久化存储
详见http://dblab.xmu.edu.cn/blog/spark-quick-start-guide/ 主要是两个步骤:
注意:
其他可能出现的错误:
[hadoop@hadoop01 bin]$ ./spark-shell
... ...
Caused by: java.net.ConnectException: Call From hadoop01/192.168.146.130 to hadoop01:9000 failed on connection exception: java.net.ConnectException: 拒绝连接;For more details see: http://wiki.apache.org/hadoop/ConnectionRefused;
... 104 more
<console>:14: error: not found: value spark
import spark.implicits._
^
<console>:14: error: not found: value spark
import spark.sql
^
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 2.2.0
/_/
Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_112)
Type in expressions to have them evaluated.
Type :help for more information.
scala>
scala> val lines = sc.textFile("/home/hadoop/look.sh")
<console>:17: error: not found: value sc
val lines = sc.textFile("/home/hadoop/look.sh")
^
hadoop dfs -chmod 777 /tmp/hive
来修改其权限,如果出现 Name node is in safe mode,那么则需要使用hadoop dfsadmin -safemode leave
来退出安全模式,之后便可以正常修改权限,改完之后再执行spark-shell变会出现正常的初始化结果:17/07/02 13:27:43 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
17/07/02 13:27:54 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
Spark context Web UI available at http://192.168.146.130:4040
Spark context available as 'sc' (master = local[*], app id = local-1498973265138).
Spark session available as 'spark'.
注意上述的三行初始化信息!
scala> val lines = sc.textFile("/home/hadoop/look.sh")
lines: org.apache.spark.rdd.RDD[String] = /home/hadoop/look.sh MapPartitionsRDD[1] at textFile at <console>:24
scala> lines.count()
org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: hdfs://hadoop01:9000/home/hadoop/look.sh
at org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:287)
at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:229)
at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:315)
at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:194)
at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:252)
at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:250)
at scala.Option.getOrElse(Option.scala:121)
at org.apache.spark.rdd.RDD.partitions(RDD.scala:250)
at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35)
at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:252)
at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:250)
at scala.Option.getOrElse(Option.scala:121)
at org.apache.spark.rdd.RDD.partitions(RDD.scala:250)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2087)
at org.apache.spark.rdd.RDD.count(RDD.scala:1158)
... 48 elided
scala> val lines = sc.textFile("file:///home/hadoop/look.sh")
lines: org.apache.spark.rdd.RDD[String] = file:///home/hadoop/look.sh MapPartitionsRDD[3] at textFile at <console>:24
scala> lines.count()
res1: Long = 26
scala> lines.first()
res2: String = #!/bin/bash
注意:
注意: 类似于Hadoop,如果开发环境不在集群内,例如在自己PC中的IDEA进行开发(使用虚拟机同理),那么就会产生两种运行方式,一是本地运行,二是提交集群运行。 本质上两种方式都是先打包,再上传(本地或集群)。即流程是一致的,但是在PC中引入的spark-core的作用是不同的,提交集群运行时,PC中的spark-core内容只是作为语法检查,类方法调用等辅助作用;但是本地运行时,除了上述功能外,其还充当了计算部分,即可以使PC成为一个类似节点的且有计算能力的存在。
全部步骤: PC上安装Scala环境,IDEA,IDEA安装Scala插件
1.本地运行
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
object WordCount extends App {
// 读取hdfs文件
val path = "hdfs://192.168.146.130:9000/spark/look.sh"
//远程调试
val conf = new SparkConf()
.setAppName("scalasparktest")
.setMaster("spark://192.168.146.130:7077")
.setJars(List("I:\\IDEA_PROJ\\ScalaSparkTest\\out\\scalasparktest_jar\\scalasparktest.jar"))
val sc = new SparkContext(conf)
val lines = sc.textFile(path)
val words = lines.flatMap(_.split(" ")).filter(word => word != " ")
val pairs = words.map(word => (word, 1))
val wordscount: RDD[(String, Int)] = pairs.reduceByKey(_ + _)
wordscount.collect.foreach(println)
}
打印结果: 注意下述的IP地址和file路径,确实是在本地运行的,而且就是引入的sparl-core起的作用
D:\Java\jdk1.8.0_77\bin\java "-javaagent:D:\JetBrains\IntelliJ IDEA
...
17/11/28 00:40:21 INFO Executor: Starting executor ID driver on host localhost
17/11/28 00:40:21 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 58570.
17/11/28 00:40:21 INFO NettyBlockTransferService: Server created on 192.168.230.1:58570
17/11/28 00:40:21 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy
17/11/28 00:40:21 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 192.168.230.1, 58570, None)
17/11/28 00:40:21 INFO BlockManagerMasterEndpoint: Registering block manager 192.168.230.1:58570 with 1992.9 MB RAM, BlockManagerId(driver, 192.168.230.1, 58570, None)
...
17/11/28 00:40:22 INFO HadoopRDD: Input split: file:/C:/Users/msi/Desktop/xiaomi2.txt:0+903
17/11/28 00:40:22 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 1111 bytes result sent to driver
17/11/28 00:40:22 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 183 ms on localhost (executor driver) (1/1)
...
(小米客服那些事,1)
(贤艾森秋t4krP0,1)
(北京IHG向,1)
17/11/28 00:40:22 INFO SparkContext: Invoking stop() from shutdown hook
17/11/28 00:40:22 INFO SparkUI: Stopped Spark web UI at http://192.168.230.1:4040
...
Process finished with exit code 0
2.提交集群运行
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
object WordCount extends App {
// 读取hdfs文件
val path = "hdfs://192.168.146.130:9000/spark/look.sh"
//远程调试
val conf = new SparkConf()
.setAppName("scalasparktest")
.setMaster("spark://192.168.146.130:7077")
.setJars(List("I:\\IDEA_PROJ\\ScalaSparkTest\\out\\scalasparktest_jar\\scalasparktest.jar"))
val sc = new SparkContext(conf)
val lines = sc.textFile(path)
val words = lines.flatMap(_.split(" ")).filter(word => word != " ")
val pairs = words.map(word => (word, 1))
val wordscount: RDD[(String, Int)] = pairs.reduceByKey(_ + _)
wordscount.collect.foreach(println)
}
此处一定要选择对Module(不是默认)和要运行的MainClass
点击OK后,选择Jar打包后的路径
使用命令: 启动master: ./sbin/start-master.sh 启动worker: ./bin/spark-class org.apache.spark.deploy.worker.Worker spark://192.168.146.130:7077 需要配置spark-env.sh中:(下面设为localhost就远程不了了) export SPARK_MASTER_HOST=192.168.146.130 export SPARK_LOCAL_IP=192.168.146.130 注意更新配置文件后需要把master和worker都重启才可以生效(单机两者都在一个机器上的情况)
出现的错误: 错误:java.io.FileNotFoundException: Jar I:\IDEA_PROJ\ScalaSparkTest\out\scalasparktest.jar not found 解决:修改setJar方法参数中的jar路径
错误:Could not connect to spark://192.168.146.130:7077 解决:重启worker和master,前提是spark-env.sh中的MASTER_IP和WORKER_IP要设置正确
错误:Exception: Call From msi-PC/192.168.230.1 to 192.168.146.130:8020 failed on connection exception: java.net.ConnectException: Connection refused: no further information; 解决:hdfs端口错误,很多教程写的是8020端口,但我hdfs是9000端口,所以要更正
错误:Invalid signature file digest for Manifest main attributes 解决:打包的文件很大,把全部依赖都打包了,90多M,但正常应该10多M,删掉无用的依赖,并且把sbt中spark-core的依赖设为provided模式
错误:重复出现如下错误
17/11/28 20:20:52 WARN TaskSchedulerImpl: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources
17/11/28 20:21:07 WARN TaskSchedulerImpl: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources
17/11/28 20:21:22 WARN TaskSchedulerImpl: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources
解决:Worker失效后被kill了[此时jps应该是没有Worker的],重启Worker即可,还不行就将hadoop和spark都重启
提交集群运行的结果:(注意IP和端口,确实是提交到集群/虚拟机 上运行后返回的结果) 整个过程全部在IDEA中,完全达到了本地调试,自动上传集群,并返回结果的流程
D:\Java\jdk1.8.0_77\bin\java "-javaagent:D:\JetBrains\IntelliJ IDEA
...
17/11/28 02:09:39 INFO StandaloneAppClient$ClientEndpoint: Executor added: app-20170630223625-0006/0 on worker-20170630215502-192.168.146.130-50762 (192.168.146.130:50762) with 1 cores
17/11/28 02:09:39 INFO StandaloneSchedulerBackend: Granted executor ID app-20170630223625-0006/0 on hostPort 192.168.146.130:50762 with 1 cores, 1024.0 MB RAM
17/11/28 02:09:39 INFO StandaloneAppClient$ClientEndpoint: Executor updated: app-20170630223625-0006/0 is now RUNNING
...
17/11/28 02:09:43 INFO BlockManagerMasterEndpoint: Registering block manager 192.168.146.130:47071 with 413.9 MB RAM, BlockManagerId(0, 192.168.146.130, 47071, None)
...
17/11/28 02:09:50 INFO TaskSchedulerImpl: Adding task set 0.0 with 2 tasks
17/11/28 02:09:50 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, 192.168.146.130, executor 0, partition 0, ANY, 4853 bytes)
...
17/11/28 02:09:55 INFO TaskSetManager: Starting task 1.0 in stage 0.0 (TID 1, 192.168.146.130, executor 0, partition 1, ANY, 4853 bytes)
...
17/11/28 02:09:55 INFO TaskSchedulerImpl: Adding task set 1.0 with 2 tasks
17/11/28 02:09:55 INFO TaskSetManager: Starting task 0.0 in stage 1.0 (TID 2, 192.168.146.130, executor 0, partition 0, NODE_LOCAL, 4625 bytes)
...
17/11/28 02:09:56 INFO TaskSetManager: Starting task 1.0 in stage 1.0 (TID 3, 192.168.146.130, executor 0, partition 1, NODE_LOCAL, 4625 bytes)
...
(-ef|grep,1)
($Jarstr,1)
([[,1)
(do,1)
(YES,1)
(while,1)
("$Jarinfo",1)
(echo,1)
(#!/bin/bash,1)
17/11/28 02:09:56 INFO SparkContext: Invoking stop() from shutdown hook
...
Process finished with exit code 0