学习自 MOOC Spark编程基础
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 2.1.0
/_/
Using Scala version 2.11.8 (OpenJDK 64-Bit Server VM, Java 1.8.0_131)
Type in expressions to have them evaluated.
Type :help for more information.
scala> val lines = sc.textFile("file:///home/hadoop/workspace/word.txt")
lines: org.apache.spark.rdd.RDD[String] = file:////home/hadoop/workspace/word.txt MapPartitionsRDD[1] at textFile at <console>:24
scala> val lines = sc.textFile("hdfs://localhost:9000/user/word.txt")
lines: org.apache.spark.rdd.RDD[String] = hdfs://localhost:9000/user/word.txt MapPartitionsRDD[3] at textFile at <console>:24
scala> val lines = sc.textFile("/user/word.txt")
lines: org.apache.spark.rdd.RDD[String] = /user/word.txt MapPartitionsRDD[9] at textFile at <console>:24
scala> val array = Array(1,2,3,4,5)
array: Array[Int] = Array(1, 2, 3, 4, 5)
scala> val rdd = sc.parallelize(array)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[12] at parallelize at <console>:26
filter(func)
,过滤scala> val linesWithSpark = lines.filter(line=>line.contains("spark"))
linesWithSpark: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[13] at filter at <console>:26
map(func)
, 映射scala> val rdd2 = rdd.map(x => x+10)
rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[14] at map at <console>:28
scala> val words = lines.map(line => line.split(" "))
words: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[15] at map at <console>:26
输出: n 个元素,每个元素是一个 String 数组
flatMap(func)
scala> val words = lines.flatMap(line => line.split(" "))
words: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[16] at flatMap at <console>:26
输出:所有单词
groupByKey(), reduceByKey(func)
按 key 合并,得到 value list,后者还可以根据 func 对 value list 进行操作spark 遇到 RDD action 时才会真正的开始执行,遇到转换的时候,只是记录下来,并不真正执行
count()
,统计 rdd 元素个数collect()
,以数组形式返回所有的元素first()
,返回第一个元素take(n)
,返回前 n 个元素reduce(func)
,聚合foreach(func)
,遍历scala> val rdd = sc.parallelize(Array(1,2,3,4,5))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24
scala> rdd.count()
res0: Long = 5
scala> rdd.first()
res1: Int = 1
scala> rdd.take(3)
res2: Array[Int] = Array(1, 2, 3)
scala> rdd.reduce((a,b)=>a+b)
res3: Int = 15
scala> rdd.collect()
res4: Array[Int] = Array(1, 2, 3, 4, 5)
scala> rdd.foreach(elem => println(elem))
persist()
,对一个 rdd 标记为持久化,遇到第一个 rdd动作 时,才真正持久化scala> val list = List("Hadoop","Spark","Hive")
list: List[String] = List(Hadoop, Spark, Hive)
scala> val rdd1 = sc.parallelize(list)
rdd1: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[1] at parallelize at <console>:26
scala> println(rdd1.count())
3
scala> println(rdd1.collect().mkString("--"))
Hadoop--Spark--Hive
scala> rdd1.cache() # 缓存起来,后续用到rdd1的时候,不用从头开始计算了
res10: rdd1.type = ParallelCollectionRDD[1] at parallelize at <console>:26
分区原则:分区个数尽量 = 集群CPU核心数
sc.textFile(path, partitionNum)
scala> val arr = Array(1,2,3,4,5)
arr: Array[Int] = Array(1, 2, 3, 4, 5)
scala> val rdd = sc.parallelize(arr, 2) # 2 个分区
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:26
scala> rdd.partitions.size
res0: Int = 2
scala> val rdd1 = rdd.repartition(1)
rdd1: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[4] at repartition at <console>:28
scala> rdd1.partitions.size
res1: Int = 1
scala> val lines = sc.
| textFile("/user/word.txt") # 读取文件
lines: org.apache.spark.rdd.RDD[String] = /user/word.txt MapPartitionsRDD[6] at textFile at <console>:25
scala> val wordCount = lines.flatMap(line => line.split(" ")).
| map(word => (word, 1)).reduceByKey((a, b) => a+b)
wordCount: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[9] at reduceByKey at <console>:27
scala> wordCount.collect() # 收集
res2: Array[(String, Int)] = Array((love,2), (spark,1), (c++,1), (i,2), (michael,1))
scala> wordCount.foreach(println) # 打印
(spark,1)
(c++,1)
(i,2)
(michael,1)
(love,2)
scala> val rdd = sc.parallelize(Array(("spark",2),("hadoop",3),("hadoop",7),("spark",3)))
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[0] at parallelize at <console>:24
scala> rdd.mapValues(x => (x, 1)).reduceByKey((x,y)=>(x._1+y._1, x._2+y._2)).mapValues(x => (x._1/x._2)).collect()
res0: Array[(String, Int)] = Array((spark,2), (hadoop,5))
scala> val textFile = sc.
| textFile("file:///home/hadoop/workspace/word.txt")
textFile: org.apache.spark.rdd.RDD[String] = file:///home/hadoop/workspace/word.txt MapPartitionsRDD[5] at textFile at <console>:25
scala> textFile.
| saveAsTextFile("file:///home/hadoop/workspace/writeword")
# 后面跟的是一个目录,而不是文件名
ls /home/hadoop/workspace/writeword/
part-00000 part-00001 _SUCCESS
hadoop@dblab-VirtualBox:/usr/local/spark/bin$ cat /home/hadoop/workspace/writeword/part-00000
i love programming
it is very interesting
scala> val textFile = sc.textFile("file:///home/hadoop/workspace/writeword")
textFile: org.apache.spark.rdd.RDD[String] = file:///home/hadoop/workspace/writeword MapPartitionsRDD[9] at textFile at <console>:24
scala> val textFile =
| sc.textFile("hdfs://localhost:9000/user/word.txt")
textFile: org.apache.spark.rdd.RDD[String] = hdfs://localhost:9000/user/word.txt MapPartitionsRDD[11] at textFile at <console>:25
scala> textFile.first()
res6: String = i love programming
保存到 hdfs (默认 当前用户的目录前缀 /user/用户名/
)
scala> textFile.saveAsTextFile("writeword")
查看 hdfs
hadoop@dblab-VirtualBox:/usr/local/hadoop/bin$ ./hdfs dfs -ls -R /user/
drwxr-xr-x - hadoop supergroup 0 2021-04-22 16:01 /user/hadoop
drwxr-xr-x - hadoop supergroup 0 2021-04-21 22:48 /user/hadoop/.sparkStaging
drwx------ - hadoop supergroup 0 2021-04-21 22:48 /user/hadoop/.sparkStaging/application_1618998320460_0002
-rw-r--r-- 1 hadoop supergroup 73189 2021-04-21 22:48 /user/hadoop/.sparkStaging/application_1618998320460_0002/__spark_conf__.zip
-rw-r--r-- 1 hadoop supergroup 120047699 2021-04-21 22:48 /user/hadoop/.sparkStaging/application_1618998320460_0002/__spark_libs__4686608713384839717.zip
drwxr-xr-x - hadoop supergroup 0 2021-04-22 16:01 /user/hadoop/writeword
-rw-r--r-- 1 hadoop supergroup 0 2021-04-22 16:01 /user/hadoop/writeword/_SUCCESS
-rw-r--r-- 1 hadoop supergroup 42 2021-04-22 16:01 /user/hadoop/writeword/part-00000
-rw-r--r-- 1 hadoop supergroup 20 2021-04-22 16:01 /user/hadoop/writeword/part-00001
drwxr-xr-x - hadoop supergroup 0 2017-11-05 21:57 /user/hive
drwxr-xr-x - hadoop supergroup 0 2017-11-05 21:57 /user/hive/warehouse
drwxr-xr-x - hadoop supergroup 0 2017-11-05 21:57 /user/hive/warehouse/hive.db
-rw-r--r-- 1 hadoop supergroup 62 2021-04-21 20:06 /user/word.txt
hadoop@dblab-VirtualBox:/usr/local/hadoop/bin$ cat /usr/local/spark/examples/src/main/resources/people.json
{"name":"Michael"}
{"name":"Andy", "age":30}
{"name":"Justin", "age":19}
scala> val jsonStr = sc.
| textFile("file:///usr/local/spark/examples/src/main/resources/people.json")
jsonStr: org.apache.spark.rdd.RDD[String] = file:///usr/local/spark/examples/src/main/resources/people.json MapPartitionsRDD[14] at textFile at <console>:25
scala> jsonStr.foreach(println)
{"name":"Michael"}
{"name":"Andy", "age":30}
{"name":"Justin", "age":19}
scala.util.parsing.json.JSON
JSON.parseFull(jsonString : String)
返回 Some or None
编写程序
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import scala.util.parsing.json.JSON
object JSONRead{
def main(args:Array[String]){
val inputFile = "file:///usr/local/spark/examples/src/main/resources/people.json"
val conf = new SparkConf().setAppName("JSONRead")
val sc = new SparkContext(conf)
val jsonStrs = sc.textFile(inputFile)
val res = jsonStrs.map(s => JSON.parseFull(s))
res.foreach({ r => r match {
case Some(map:Map[String, Any]) => println(map)
case None => println("parsing failed")
case other => println("unknown data structure: " + other
)}}
)
}
}
使用 sbt 编译打包为 jar,spark-submit --class "JSONRead" <路径 of jar>
(有待实践操作)
参考: 使用Intellij Idea编写Spark应用程序(Scala+SBT) http://dblab.xmu.edu.cn/blog/1492-2/
hadoop@dblab-VirtualBox:/usr/local/hbase/bin$ ./hbase shell
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/usr/local/hbase/lib/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/usr/local/hadoop/share/hadoop/common/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
HBase Shell; enter 'help<RETURN>' for list of supported commands.
Type "exit<RETURN>" to leave the HBase Shell
Version 1.1.5, r239b80456118175b340b2e562a5568b5c744252e, Sun May 8 20:29:26 PDT 2016
hbase(main):001:0> disable "student"
0 row(s) in 3.0730 seconds
hbase(main):002:0> drop "student"
0 row(s) in 1.3530 seconds
hbase(main):003:0> create "student","info"
0 row(s) in 1.3570 seconds
=> Hbase::Table - student
hbase(main):004:0> put "student","1","info:name","michael"
0 row(s) in 0.0920 seconds
hbase(main):005:0> put "student","1","info:gender","M"
0 row(s) in 0.0410 seconds
hbase(main):006:0> put "student","1","info:age","18"
0 row(s) in 0.0080 seconds
也需要编写程序,sbt 编译打包