// Scala
import org.apache.spark.sql.SparkSession
val sparkSession = SparkSession \
    .builder() \
    .appName("myApp") \
    .enableHiveSupport() \
    .getOrCreate()
! // Two ways you can access spark context from spark session
val spark_context = sparkSession._sc
val spark_context = sparkSession.sparkContext

# PySpark
from pyspark.sql import SparkSession
spark_session = SparkSession \
    .builder \
    .enableHiveSupport() \
    .getOrCreate()
# Two ways you can access spark context from spark session
spark_context = spark_session._sc
spark_context = spark_session.sparkContext

scala> spark
res16: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@3b5a4aa5
+ // spark package contains sql package and SparkContext class (instantiation as "sc" in spark shell) SparkContext sql package contains SparkSession  
+ // class (instantiation as "spark"). SparkSession class contains sparkContext class. One is SparkContext, the other is sparkContext. 

scala> spark.
baseRelationToDataFrame   createDataFrame   experimental      range          sharedState    stop      udf       
catalog                   createDataset     implicits         read           sparkContext   streams   version   
close                     emptyDataFrame    listenerManager   readStream     sql            table               
conf                      emptyDataset      newSession        sessionState   sqlContext     time                
// SparkSession is a class belongs to org.apache.spark.sql package

scala> spark.sparkContext
res14: org.apache.spark.SparkContext = org.apache.spark.SparkContext@318ff889

scala> sc
res15: org.apache.spark.SparkContext = org.apache.spark.SparkContext@318ff889

// Create an String Array
scala> val carsArray = Array[String]("BMW", "Bentley", "Mercedes", "Suzuki", "Honda", "Jaguar", "Fiat", "Audi")
carsArray: Array[String] = Array(BMW, Bentley, Mercedes, Suzuki, Honda, Jaguar, Fiat, Audi)
!// Create RDD from the Array (Array is data structure in scala/java it's not distributed until being converted to RDD data type, then it has features
!// of partitions, fault tolerence, immutable, lazy evaluation and so on)
scala> val carsRDD = spark.sparkContext.parallelize(carsArray, 2)
carsRDD: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[8] at parallelize at <console>:25
@@// we can also use sc instead of spark.sparkContext@@
scala> carsRDD.collect()
res17: Array[String] = Array(BMW, Bentley, Mercedes, Suzuki, Honda, Jaguar, Fiat, Audi)

-// duplicates of BMW in the array
scala> val carsArray = Array[String]("BMW", "Bentley", "Mercedes", "Suzuki", "Honda", "Jaguar", "Fiat", "Audi", "BMW")
carsArray: Array[String] = Array(BMW, Bentley, Mercedes, Suzuki, Honda, Jaguar, Fiat, Audi, BMW)

scala> val carsRDD = sc.parallelize(carsArray)
carsRDD: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[9] at parallelize at <console>:26

scala> carsRDD.collect()
res19: Array[String] = Array(BMW, Bentley, Mercedes, Suzuki, Honda, Jaguar, Fiat, Audi, BMW)
+// apply distinct() method to get distinct values in RDD
scala> val distinctcarsRDD = carsRDD.distinct()
distinctcarsRDD: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[12] at distinct at <console>:25

scala> distinctcarsRDD.collect()
res20: Array[String] = Array(Bentley, Suzuki, BMW, Mercedes, Jaguar, Fiat, Honda, Audi)

scala> val carsArray = Array[String]("BMW", "Bentley", "Mercedes", "Suzuki", "Honda", "Jaguar", "Fiat", "Audi")
carsArray: Array[String] = Array(BMW, Bentley, Mercedes, Suzuki, Honda, Jaguar, Fiat, Audi)

scala> val carsRDD = sc.parallelize(carsArray)
carsRDD: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[13] at parallelize at <console>:26

scala> carsRDD.collect()
res21: Array[String] = Array(BMW, Bentley, Mercedes, Suzuki, Honda, Jaguar, Fiat, Audi)
+//startsWith() is one of the methods for string in Scala. 
scala> carsRDD.filter(carName => carName.startsWith("B")).collect()
res22: Array[String] = Array(BMW, Bentley)

scala> val numbersRDD = sc.parallelize(1 to 10, 2)
numbersRDD: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[25] at parallelize at <console>:24

scala> numbersRDD.filter(x => x%2 ==0).collect()
res31: Array[Int] = Array(2, 4, 6, 8, 10)

+//shorthand notation
scala> numbersRDD.filter(_%2 ==0).collect()
res32: Array[Int] = Array(2, 4, 6, 8, 10)

+ //define a function to get even numbers, the return type is Boolean
scala> def evenFilter(x : Int) =
     | x%2 == 0
evenFilter: (x: Int)Boolean

scala> numbersRDD.collect()
res34: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

scala> val evennumbers = numbersRDD.filter(x => evenFilter(x))
evennumbers: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[28] at filter at <console>:27

scala> evennumbers.collect()
res35: Array[Int] = Array(2, 4, 6, 8, 10)

+//use function instead of logic definition
scala> val evennumbers = numbersRDD.filter(evenFilter).collect()
evennumbers: Array[Int] = Array(2, 4, 6, 8, 10)

scala> carsRDD.collect()
res38: Array[String] = Array(BMW, Bentley, Mercedes, Suzuki, Honda, Jaguar, Fiat, Audi)

scala> val carsRDDwithB = carsRDD.map(carName => (carName, carName.startsWith("B")))
carsRDDwithB: org.apache.spark.rdd.RDD[(String, Boolean)] = MapPartitionsRDD[30] at map at <console>:25

+// Make a combination of data for each RDD elements
scala> carsRDDwithB.collect()
res39: Array[(String, Boolean)] = Array((BMW,true), (Bentley,true), (Mercedes,false), (Suzuki,false), (Honda,false), (Jaguar,false), (Fiat,false), (Audi,false))

+// Map transformation to get (CarName + CarName length)
scala> val carsRDDMap = carsRDD.map(carName => (carName, carName.length)).collect()
carsRDDMap: Array[(String, Int)] = Array((BMW,3), (Bentley,7), (Mercedes,8), (Suzuki,6), (Honda,5), (Jaguar,6), (Fiat,4), (Audi,4))

scala> val array1D = Array("1,2,3", "4,5,6", "7,8,9")
array1D: Array[String] = Array(1,2,3, 4,5,6, 7,8,9)

+//map()
scala> val array2D = array1D.map(x => x.split(","))
array2D: Array[Array[String]] = Array(Array(1, 2, 3), Array(4, 5, 6), Array(7, 8, 9))
+//flatMap() is combination of map() and flatten()
scala> val FlatArray = array1D.flatMap(x => x.split("x"))
FlatArray: Array[String] = Array(1,2,3, 4,5,6, 7,8,9)

scala> val FlatArray = array1D.flatMap(x => x.split(","))
FlatArray: Array[String] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)

// Create an Array of texts
scala> val book = Array("A book is a medium for recording information in the form of writing or images, typically composed of many pages (made of papyrus, parchment, vellum, or paper) bound together and protected by a cover.", "The technical term for this physical arrangement is codex (plural, codices). In the history of hand-held physical supports for extended written compositions or records, the codex replaces its predecessor, the scroll. A single sheet in a codex is a leaf and each side of a leaf is a page.", "As an intellectual object, a book is prototypically a composition of such great length that it takes a considerable investment of time to compose and still considered as an investment of time to read.", "In a restricted sense, a book is a self-sufficient section or part of a longer composition, a usage reflecting that, in antiquity, long works had to be written on several scrolls and each scroll had to be identified by the book it contained.", "Each part of Aristotle's Physics is called a book. In an unrestricted sense, a book is the compositional whole of which such sections, whether called books or chapters or parts, are parts.")
book: Array[String] = Array(A book is a medium for recording information in the form of writing or images, typically composed of many pages (made of papyrus, parchment, vellum, or paper) bound together and protected by a cover., The technical term for this physical arrangement is codex (plural, codices). In the history of hand-held physical supports for extended written compositions or records, the codex replaces its predecessor, the scroll. A single sheet in a codex is a leaf and each side of a leaf is a page., As an intellectual object, a book is prototypically a composition of such great length that it takes a considerable investment of time to compose and still considered as an investment of time to read., In a restricted sense, a book is a self-sufficient section or part of a longe...

+//Create the RDD by parallelizing an existing collection in your driver program                        
scala> val bookRDD = spark.sparkContext.parallelize(book, 2)
bookRDD: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[33] at parallelize at <console>:25

+//Split all the texts in the array into words 
scala> val flatbookRDD = bookRDD.flatMap(text => text.split(" "))
flatbookRDD: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[34] at flatMap at <console>:25

scala> flatbookRDD.sortBy(word => word.length())
res48: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[39] at sortBy at <console>:26
//Ascending
scala> flatbookRDD.sortBy(word => word.length()).collect()
res49: Array[String] = Array(a, A, a, a, a, a, a, a, a, a, a, a, a, a, of, by, In, of, or, in, is, of, is, As, an, is, of, it, of, to, as, an, of, to, In, is, or, of, in, to, be, on, to, be, by, it, In, an, is, of, or, or, for, the, and, The, for, the, for, the, its, and, had, and, had, the, the, are, form, term, this, each, side, leaf, book, such, that, time, time, book, part, long, each, book, Each, part, book, such, codex, sheet, codex, great, takes, still, read., usage, that,, works, book., whole, which, books, medium, single, length, sense,, longer, scroll, called, sense,, called, parts,, parts., history, written, a page., object,, compose, section, written, several, scrolls, whether, composed, a cover., physical, physical, supports, extended, records,, replaces, chapters, A book i...
scala> 
//Descending 
scala> flatbookRDD.sortBy(word => word.length(), false).collect()
res50: Array[String] = Array(of papyrus, parchment, vellum,, is codex (plural, codices)., of Aristotle's Physics is, recording information in, or paper) bound together, of writing or images,, many pages (made, self-sufficient, prototypically, compositional, compositions, predecessor,, intellectual, considerable, composition,, unrestricted, arrangement, the scroll., composition, a leaf and, investment, considered, investment, restricted, reflecting, antiquity,, identified, contained., A book is, typically, protected, technical, hand-held, sections,, composed, a cover., physical, physical, supports, extended, records,, replaces, chapters, history, written, a page., object,, compose, section, written, several, scrolls, whether, medium, single, length, sense,, longer, scroll, called, sense,...

//Random split the RDD by 70%:30% into Array[RDD[String]]
scala> val MLbookRDD = flatbookRDD.randomSplit(Array[Double](0.7,0.3))
MLbookRDD: Array[org.apache.spark.rdd.RDD[String]] = Array(MapPartitionsRDD[52] at randomSplit at <console>:25, MapPartitionsRDD[53] at randomSplit at <console>:25)
+//Call the first element of the RDD Array, which makes up 70% of the RDD for training purpose
scala> MLbookRDD(0).collect()
res54: Array[String] = Array(A book is, a, medium, for, form, of writing or images,, typically, many pages (made, or paper) bound together, and, protected, a cover., The, technical, term, for, physical, arrangement, the, of, hand-held, for, extended, written, or, records,, the, replaces, its, predecessor,, the scroll., single, in, is, a leaf and, each, a, leaf, is, a page., As, an, intellectual, object,, a, is, a, composition, of, great, length, that, it, takes, a, considerable, investment, of, time, to, and, still, investment, to, restricted, sense,, a, book, is, a, section, or, part, of, a, longer, composition,, a, that,, in, antiquity,, long, had, be, on, several, scrolls, scroll, to, be, identified, by, the, book, it, Each, part, called, a, In, unrestricted, sense,, book, is, the, w...
+//Call the second element of the RDD Array, which makes up 30% of the RDD for test purpose
scala> MLbookRDD(1).collect()
res55: Array[String] = Array(recording information in, the, composed, of, of papyrus, parchment, vellum,, by, this, is codex (plural, codices)., In, history, physical, supports, compositions, codex, A, sheet, a, codex, side, of, book, prototypically, such, compose, considered, as, an, of, time, read., In, a, self-sufficient, usage, reflecting, works, to, written, and, each, had, contained., of Aristotle's Physics is, book., an, a, compositional, such, chapters)

+//Display all the elements in this RDD (from all partitions)
scala> flatbookRDD.collect()
res0: Array[String] = Array(A book is, a, medium, for, recording information in, the, form, of writing or images,, typically, composed, of, many pages (made, of papyrus, parchment, vellum,, or paper) bound together, and, protected, by, a cover., The, technical, term, for, this, physical, arrangement, is codex (plural, codices)., In, the, history, of, hand-held, physical, supports, for, extended, written, compositions, or, records,, the, codex, replaces, its, predecessor,, the scroll., A, single, sheet, in, a, codex, is, a leaf and, each, side, of, a, leaf, is, a page., As, an, intellectual, object,, a, book, is, prototypically, a, composition, of, such, great, length, that, it, takes, a, considerable, investment, of, time, to, compose, and, still, considered, as, an, investment, of, tim...

scala> sc.parallelize(1 to 10).collect()
res2: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

scala> sc.parallelize(1 to 10).reduce(_ + _)
res3: Int = 55

+//define a logic function taking 2 arguments (word here) and return the word with smaller length 
scala> def smallestWord(leftWord:String, rightWord:String):String={
     | if (leftWord.length >= rightWord.length) {rightWord}
     | else {leftWord}}
smallestWord: (leftWord: String, rightWord: String)String

//use the self-defined function as the reduce behavior
scala> flatbookRDD.reduce(smallestWord)
res9: String = a

//without using the pre-defined logic function (NOT RECCOMENDED)
scala> flatbookRDD.reduce((x,y)=>{if(x.length>=y.length) {y} else {x}})
res10: String = a

scala> flatbookRDD.count()
res11: Long = 166

// Returns a potentially incomplete result (range from 105 to 136) in 16 millisecond timeout. Default confidence interval is 95%.
scala> flatbookRDD.countApprox(16)
res26: org.apache.spark.partial.PartialResult[org.apache.spark.partial.BoundedDouble] = (partial: [105.000, 136.000])

// Count returned within 20 millisecond timeout
scala> flatbookRDD.countApprox(20)
res12: org.apache.spark.partial.PartialResult[org.apache.spark.partial.BoundedDouble] = (final: [166.000, 166.000])

scala> flatbookRDD.countByValue()
res28: scala.collection.Map[String,Long] = Map(side -> 1, parts. -> 1, for -> 3, works -> 1, Each -> 1, records, -> 1, this -> 1, in -> 2, scroll -> 1, are -> 1, is -> 5, protected -> 1, extended -> 1, compose -> 1, history -> 1, predecessor, -> 1, of papyrus, parchment, vellum, -> 1, In -> 3, whether -> 1, usage -> 1, contained. -> 1, of writing or images, -> 1, a page. -> 1, book -> 4, had -> 2, antiquity, -> 1, of Aristotle's Physics is -> 1, it -> 2, prototypically -> 1, sheet -> 1, A -> 1, term -> 1, a -> 13, self-sufficient -> 1, recording information in -> 1, as -> 1, scrolls -> 1, chapters -> 1, or -> 4, such -> 2, medium -> 1, each -> 2, sections, -> 1, section -> 1, compositional -> 1, that -> 1, to -> 4, the scroll. -> 1, hand-held -> 1, single -> 1, considerable -> 1, long -...

scala> val sampleRDD = spark.sparkContext.parallelize(List(6, 7, 5, 3, 10, 25, 8, 70, 35))
sampleRDD: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[9] at parallelize at <console>:23

+//take first 3 elements from the collection
scala> sampleRDD.take(3)
res32: Array[Int] = Array(6, 7, 5)
+//take first 3 elements in Ascending Order
scala> sampleRDD.takeOrdered(3)
res33: Array[Int] = Array(3, 5, 6)
+//take first 3 elements in Descending Order
scala> sampleRDD.top(3)
res34: Array[Int] = Array(70, 35, 25)

scala> flatbookRDD.takeSample(true, 70, 10L)
res35: Array[String] = Array(books, written, In, by, scrolls, A book is, antiquity,, part, a leaf and, great, In, this, for, the scroll., section, is codex (plural, codices)., Each, term, the, object,, supports, each, technical, written, of writing or images,, chapters, In, a, sense,, a page., parts,, codex, that,, self-sufficient, extended, written, The, protected, and, this, compositional, length, investment, book., composed, composition,, a, or paper) bound together, of, book, a, its, a, a, part, an, A book is, compose, supports, composed, a, leaf, its, In, of, or, by, physical, for, records,)

scala> flatbookRDD.saveAsTextFile("/home/tony/temp20220508")

@@tony@tony-ubuntu:~/temp20220508$ hdfs dfs -ls /home/tony/temp20220508@@
Found 3 items
-rw-r--r--   1 tony supergroup          0 2022-05-08 01:05 /home/tony/temp20220508/_SUCCESS
-rw-r--r--   1 tony supergroup        511 2022-05-08 01:05 /home/tony/temp20220508/part-00000
-rw-r--r--   1 tony supergroup        635 2022-05-08 01:05 /home/tony/temp20220508/part-00001

scala> val flatbookRDD = bookRDD.flatMap(x => x.split(" "))
flatbookRDD: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[18] at flatMap at <console>:25

scala> flatbookRDD.take(5)
res42: Array[String] = Array(A book is, a, medium, for, recording information in)

scala> flatbookRDD.top(5)
res43: Array[String] = Array(written, written, works, whole, which)

scala> val flatbookRDD = bookRDD.flatMap(x => x.split(" "))
flatbookRDD: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[20] at flatMap at <console>:25

+//cache RDD in Memory 
scala> flatbookRDD.cache()
res46: flatbookRDD.type = MapPartitionsRDD[20] at flatMap at <console>:25

scala> flatbookRDD.take(5)
res47: Array[String] = Array(A book is, a, medium, for, recording information in)

+// will not execute flatMap() tranformation again, intead will take from memory.
scala> flatbookRDD.top(5)
res48: Array[String] = Array(written, written, works, whole, which)

scala> flatbookRDD.getStorageLevel
+res49: org.apache.spark.storage.StorageLevel = StorageLevel(memory, deserialized, 1 replicas)

-//uncache or unpersist the RDD storage level since it's already being assigned in_memory by using cache() above.
scala> flatbookRDD.unpersist()
res51: flatbookRDD.type = MapPartitionsRDD[20] at flatMap at <console>:25

+//persist() to assign the RDD storage level to disk only
scala> flatbookRDD.persist(org.apache.spark.storage.StorageLevel.DISK_ONLY)
res52: flatbookRDD.type = MapPartitionsRDD[20] at flatMap at <console>:25
+//check storage level
scala> flatbookRDD.getStorageLevel
res53: org.apache.spark.storage.StorageLevel = StorageLevel(disk, 1 replicas)

//flatbookRDD
scala> flatbookRDD.collect()
res2: Array[String] = Array(A book is, a, medium, for, recording information in, the, form, of writing or images,, typically, composed, of, many pages (made, of papyrus, parchment, vellum,, or paper) bound together, and, protected, by, a cover., The, technical, term, for, this, physical, arrangement, is codex (plural, codices)., In, the, history, of, hand-held, physical, supports, for, extended, written, compositions, or, records,, the, codex, replaces, its, predecessor,, the scroll., A, single, sheet, in, a, codex, is, a leaf and, each, side, of, a, leaf, is, a page., As, an, intellectual, object,, a, book, is, prototypically, a, composition, of, such, great, length, that, it, takes, a, considerable, investment, of, time, to, compose, and, still, considered, as, an, investment, of, tim...
//count of the RDD, which is the number of words (1 word per line) is 166
scala> flatbookRDD.count()
res3: Long = 166

// flatbookRDD with 2 partitions, each partition returns the number of the lines (60 and 106)
scala> flatbookRDD.pipe("wc -l").collect()
res4: Array[String] = Array(60, 106)

+// Repartiton to 3 partitions
scala> val flatBookRDD = flatbookRDD.repartition(3)
flatBookRDD: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[10] at repartition at <console>:25

scala> flatBookRDD.pipe("wc -l").collect()
res12: Array[String] = Array(55, 55, 56)

// for each partition create an array element 
scala> bookRDD.mapPartitions(part => Iterator[Int](1)).collect()
res67: Array[Int] = Array(1, 1)
// thus count the number of partitons indirectly by using sum() method
scala> bookRDD.mapPartitions(part => Iterator[Int](1)).sum()
res66: Double = 2.0

// to get the number of RDD in each partition
scala> bookRDD.mapPartitions(part => Array(part.size).iterator).collect()
res79: Array[Int] = Array(2, 3)

// to display the rdd element in the second partition of the bookRDD rdd.
scala> bookRDD.mapPartitionsWithIndex((index:Int, it: Iterator[String]) => it.toArray.map(x => if (index==1){println(x)}).iterator).collect()
As an intellectual object, a book is prototypically a composition of such great length that it takes a considerable investment of time to compose and still considered as an investment of time to read.
In a restricted sense, a book is a self-sufficient section or part of a longer composition, a usage reflecting that, in antiquity, long works had to be written on several scrolls and each scroll had to be identified by the book it contained.
Each part of Aristotle's Physics is called a book. In an unrestricted sense, a book is the compositional whole of which such sections, whether called books or chapters or parts, are parts.
res4: Array[Unit] = Array((), (), (), (), ())

// print the list for each partition 
scala> bookRDD.foreachPartition((it: Iterator[String])=>println(it.toList))
List(A book is a medium for recording information in the form of writing or images, typically composed of many pages (made of papyrus, parchment, vellum, or paper) bound together and protected by a cover., The technical term for this physical arrangement is codex (plural, codices). In the history of hand-held physical supports for extended written compositions or records, the codex replaces its predecessor, the scroll. A single sheet in a codex is a leaf and each side of a leaf is a page.)
List(As an intellectual object, a book is prototypically a composition of such great length that it takes a considerable investment of time to compose and still considered as an investment of time to read., In a restricted sense, a book is a self-sufficient section or part of a longer composition, a usage reflecting that, in antiquity, long works had to be written on several scrolls and each scroll had to be identified by the book it contained., Each part of Aristotle's Physics is called a book. In an unrestricted sense, a book is the compositional whole of which such sections, whether called books or chapters or parts, are parts.)

scala> flatbookRDD.map(word => (word, word.toLowerCase)).collect()
res51: Array[(String, String)] = Array((A,a), (book,book), (is,is), (a,a), (medium,medium), (for,for), (recording,recording), (information,information), (in,in), (the,the), (form,form), (of,of), (writing,writing), (or,or), (images,,images,), (typically,typically), (composed,composed), (of,of), (many,many), (pages,pages), ((made,(made), (of,of), (papyrus,,papyrus,), (parchment,,parchment,), (vellum,,vellum,), (or,or), (paper),paper)), (bound,bound), (together,together), (and,and), (protected,protected), (by,by), (a,a), (cover.,cover.), (The,the), (technical,technical), (term,term), (for,for), (this,this), (physical,physical), (arrangement,arrangement), (is,is), (codex,codex), ((plural,,(plural,), (codices).,codices).), (In,in), (the,the), (history,history), (of,of), (hand-held,hand-held)...

+// use word length as the key to create key-value pair, value corresponds to word itself.
scala> flatbookRDD.keyBy(word => word.length).collect
res53: Array[(Int, String)] = Array((1,A), (4,book), (2,is), (1,a), (6,medium), (3,for), (9,recording), (11,information), (2,in), (3,the), (4,form), (2,of), (7,writing), (2,or), (7,images,), (9,typically), (8,composed), (2,of), (4,many), (5,pages), (5,(made), (2,of), (8,papyrus,), (10,parchment,), (7,vellum,), (2,or), (6,paper)), (5,bound), (8,together), (3,and), (9,protected), (2,by), (1,a), (6,cover.), (3,The), (9,technical), (4,term), (3,for), (4,this), (8,physical), (11,arrangement), (2,is), (5,codex), (8,(plural,), (9,codices).), (2,In), (3,the), (7,history), (2,of), (9,hand-held), (8,physical), (8,supports), (3,for), (8,extended), (7,written), (12,compositions), (2,or), (8,records,), (3,the), (5,codex), (8,replaces), (3,its), (12,predecessor,), (3,the), (7,scroll.), (1,A), (6,sing...

+scala> flatbookRDD.keyBy(word => word.length).mapValues(word => word.toUpperCase).collect
res56: Array[(Int, String)] = Array((1,A), (4,BOOK), (2,IS), (1,A), (6,MEDIUM), (3,FOR), (9,RECORDING), (11,INFORMATION), (2,IN), (3,THE), (4,FORM), (2,OF), (7,WRITING), (2,OR), (7,IMAGES,), (9,TYPICALLY), (8,COMPOSED), (2,OF), (4,MANY), (5,PAGES), (5,(MADE), (2,OF), (8,PAPYRUS,), (10,PARCHMENT,), (7,VELLUM,), (2,OR), (6,PAPER)), (5,BOUND), (8,TOGETHER), (3,AND), (9,PROTECTED), (2,BY), (1,A), (6,COVER.), (3,THE), (9,TECHNICAL), (4,TERM), (3,FOR), (4,THIS), (8,PHYSICAL), (11,ARRANGEMENT), (2,IS), (5,CODEX), (8,(PLURAL,), (9,CODICES).), (2,IN), (3,THE), (7,HISTORY), (2,OF), (9,HAND-HELD), (8,PHYSICAL), (8,SUPPORTS), (3,FOR), (8,EXTENDED), (7,WRITTEN), (12,COMPOSITIONS), (2,OR), (8,RECORDS,), (3,THE), (5,CODEX), (8,REPLACES), (3,ITS), (12,PREDECESSOR,), (3,THE), (7,SCROLL.), (1,A), (6,SING...

+//类似于mapValues，不同的在于flatMapValues应用于元素为KV对的RDD中Value。每个一元素的Value被输入函数映射为一系列的值，然后这些值再与原RDD中的Key组成一系列新的KV对
scala> flatbookRDD.keyBy(word => word.length).flatMapValues(word => word.toUpperCase).collect
res57: Array[(Int, Char)] = Array((1,A), (4,B), (4,O), (4,O), (4,K), (2,I), (2,S), (1,A), (6,M), (6,E), (6,D), (6,I), (6,U), (6,M), (3,F), (3,O), (3,R), (9,R), (9,E), (9,C), (9,O), (9,R), (9,D), (9,I), (9,N), (9,G), (11,I), (11,N), (11,F), (11,O), (11,R), (11,M), (11,A), (11,T), (11,I), (11,O), (11,N), (2,I), (2,N), (3,T), (3,H), (3,E), (4,F), (4,O), (4,R), (4,M), (2,O), (2,F), (7,W), (7,R), (7,I), (7,T), (7,I), (7,N), (7,G), (2,O), (2,R), (7,I), (7,M), (7,A), (7,G), (7,E), (7,S), (7,,), (9,T), (9,Y), (9,P), (9,I), (9,C), (9,A), (9,L), (9,L), (9,Y), (8,C), (8,O), (8,M), (8,P), (8,O), (8,S), (8,E), (8,D), (2,O), (2,F), (4,M), (4,A), (4,N), (4,Y), (5,P), (5,A), (5,G), (5,E), (5,S), (5,(), (5,M), (5,A), (5,D), (5,E), (2,O), (2,F), (8,P), (8,A), (8,P), (8,Y), (8,R), (8,U), (8,S), (8,,), (10...