一、SparkContext、SparkConf内容回顾

二、RDD创建的两种方式

• 2.1、官网对于SparkConf的描述
• 2.2、spark-submit的最佳实践

三、使用idea构建SparkContext

• 3.1、spark-shell查看命令帮助&&在客户端中使用
• 3.2、此处涉及的参数调优点

四、一些思考题

一、SparkContext、SparkConf内容回顾

1、SparkContext、SparkConf、spark-shell中一些常用参数介绍

2、开发程序前首先要构建SparkContext，构建SparkContext前要先构建SparkConf，在SparkConf中设置与spark程序相关的：appName、Master；

 //如下就是SparkConf的链式编程val sparkconf = new SparkConf().setAppName("SparkContextApp").setMaster("local[2]")//创建SparkContext前需要创建一个SparkConfval sc = new SparkContext(sparkconf)

3、spark-submit提交的时候不要使用硬编码的方式（比如每天运行的spark应用程序，如何通过作业名称知道它跑的是哪个应用程序，正常是在shell中通过时间把他拼接上去）

4、每一个Spark应用程序中都有一个SparkContext，在Spark使用过程中提供给我们的shell脚本就是在$SPARK_HOME/bin目录下的spark-shell，spark-shell就是一个spark application

5、Spark shell的启动过程中会给我们创建一个SparkContext，它的别名叫sc

6、spark-shell、spark-sql底层都是通过spark-submit的方式进行提交的。

二、RDD创建的两种方式

• Spark revolves around the concept of a resilient distributed dataset (RDD), which is a fault-tolerant collection of elements that can be operated on in parallel. There are two ways to create RDDs: parallelizing an existing collection in your driver program, or referencing a dataset in an external storage system, such as a shared filesystem, HDFS, HBase, or any data source offering a Hadoop InputFormat.

2.1、通过parallelize方法创建RDD

• parallelize是在SparkContext.scala中的方法：

  def parallelize[T: ClassTag](seq: Seq[T],numSlices: Int = defaultParallelism): RDD[T] = withScope {assertNotStopped()new ParallelCollectionRDD[T](this, seq, numSlices, Map[Int, Seq[String]]())}parallelize方法介绍：
两个参数：第一个参数是sequence的集合，第二个参数numslices：Int类型，它有个默认的值，
传递使用的时候这个参数是可以不用传递的；所以parallelize方法只需要传一个seq就行了。

• parallelize方法介绍：Distribute a local Scala collection to form an RDD, 将一个本地的scala集合转换成RDD.

在spark-shell中实操：

1、定义一个数组：
scala> val data = Array(1,2,3,4,5)
data: Array[Int] = Array(1, 2, 3, 4, 5)

2、parallelize的方式：
scala> val distData = sc.parallelize(data)
distData: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at :26

3、scala> distData.collect
res0: Array[Int] = Array(1, 2, 3, 4, 5)
使用collect进行触发，观察UI界面，发现只有2个task：

观察UI界面，发现task的数量是2，问题：2的由来？

1、once created, the distributed dataset can be operated on in parallel. （一旦被创建能够以并行化的方式被操作）

2、one important parameter for parallel collections is the number of partitions to cut the dataset into. Spark will run one task for each partition of the cluster.

• 我们设置的Partition数量会将我们的数据集拆分成number of partiiton.

3、Spark will run one task for each partition of the cluster（Spark将会运行一个task对应一个Partition），我们设置了5个partition，所以是5个task.
在Spark中，partition的数量默认就是等于task的数量，partition=task

scala> val data = Array(1,2,3,4,5)
data: Array[Int] = Array(1, 2, 3, 4, 5)scala> val distData=sc.parallelize(data,5)
distData: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:26scala> distData.collect
res2: Array[Int] = Array(1, 2, 3, 4, 5)

Typically（典型的）， you want 2-4 partitions for each CPU in your cluster. why？

• 如果一个core对应一个task，一个core设置2-4个partition，避免：一个作业跑完另一个接着跑，作业量大小不一导致资源产生过多的浪费。

Spark会尝试设置分区数基于你的集群，比如我们处理HDFS的文件（文件>BlockSize(128M)和文件大小>BlockSize(128M)。Note：task数量如果多的话，输出的小文件也会很多；就会涉及到小文件的合并了。

2.2、RDD的第二种创建方式External Datasets(外部数据集)

1、Spark can create distributed datasets from any storage source supported by Hadoop, including your system, HDFS, Cassandra, HBase, Amazon S3, Spark supports text files, and any hadoop inputFormat.

• Spark能够创建分布式的数据集从任何支持Hadoop的存储源，比如你的操作系统，HDFS/Cassandra/HBase/Amazon S3，这些文件系统上的文件都能直接使用Spark创建，支持格式：文本格式、sequencefile

Spark支不支持读取ORC格式的文件，当然是支持的

2、Text file can be created using SparkContext’s textFIle method. this method takes the URI for the file(either a local path on the machine, or a hdfs 😕/，s3n://，etc URI) and reads it as a collection of lines.

SparkContext中的textFile方法：
概念：

Read a text file from HDFS, a local file system（available on all nodes）,or Hadoop-supported file system URI, and return it as an RDD of String 读一个普通的文本文件从hdfs上，如果文件是本地系统上的话，一定要能保证在任意节点能够访问到它在SparkContext.scala中定义了一个textFile方法/*** Read a text file from HDFS, a local file system (available on all nodes), or any* Hadoop-supported file system URI, and return it as an RDD of Strings.* @param path path to the text file on a supported file system* @param minPartitions suggested minimum number of partitions for the resulting RDD* @return RDD of lines of the text file*/def textFile(path: String,minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {assertNotStopped()hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],minPartitions).map(pair => pair._2.toString).setName(path)}

问题：如何理解本地文件系统要在所有节点被访问到？
比如你搭建了一个Standalone模式，有1个master+100Worker --> wc inputsource(local)
因为采用的是分布式计算框架，不确定会把你的作业调到哪个节点去运行；如果分布式计算框架跑作业的那个节点没有inputsource，那么必然会报file not found这个错误。

建议：
如果是集群，跑hdfs上的文件；本地少用standalone测试，因为不知道会在那台机器上执行（对于本地文件）。

2.3、分别读取linux本地和hdfs上的文件

1、尝试读取linux本地上的文件：
-我在linux本地目录下准备了一份数据：/home/hadoop/data/ruozeinput.txt：

1、ruozeinput.txt我准备了如下数据在里面：
hello	hello	hello
world	world
john2、尝试读取文件：
scala> val distFile = sc.textFile("file:///home/hadoop/data/ruozeinput.txt")
distFile: org.apache.spark.rdd.RDD[String] = file:///home/hadoop/data/ruozeinput.txt MapPartitionsRDD[1] at textFile at <console>:24
//本地的文件使用sc.textFile读取进来后，返回了一个RDD，这个RDD是string类型- 如上我们已经创建完成：
once created, The distributed dataset(distdata) can be operated on in parallel, for example, we might call distFile.reduce((a,b) =>a+b) to add up the elements of the Array.3、通过collect触发action：
scala> distFile.collect
res0: Array[String] = Array(hello       hello   hello, world    world, john)- Once created,distFile can be acted on by dataset operations. For example, we can add up the sizes of all the lines using the map and reduce operations as follows:scala> distFile.map(s=>s.length).reduce((a,b)=>a+b)
res3: Int = 32//计算的是字符长度
//返回的是一个String类型，验证了这一点：reads it as a collection of lines, ruozeinput.txt中是有3行，所以这个集合有两个逗号分隔开

2、在hdfs上读取文件：

scala> val distfile = sc.textFile("hdfs://10.0.0.135/wordcount/input/ruozeinput.txt")
distfile: org.apache.spark.rdd.RDD[String] = hdfs://10.0.0.135/wordcount/input/ruozeinput.txt MapPartitionsRDD[6] at textFile at <console>:24

2.4、RDD创建的注意事项

1、 if using a path on the local filesystem, the file must also be accessible at the same path on worker nodes. Either copy the file to all workers or use a network-mounted shared file system.

• 如果这个文件在本地系统上：“file://home/hadoop/data/*.txt”，这个文件必须在各个节点上被访问的到；或者拷贝文件到所有的worker上或者使用挂载系统

2、 All of Spark’s fil-based input methods, including textFile, support running on directories, compressed files, and wildcards（通配符） as well. For example, you can use textFile("/my/directory"), textFile("/my/directory/.txt") and textFile("/my/direcctory/*.gz").

• 所有内置的方法，比如textFile既能读取文件、文件夹、压缩目录

我执行的时候还报了个错：
java.net.ConnectException: Call From hadoop004/192.168.179.155 to hadoop004:8020 failed on connection exception: java.net.ConnectException: Connection refused;主要原因是默认的是8020端口，修改成了9000端口后，读取文件要带上端口号：
1、尝试读取文件：
scala> val a=sc.textFile("hdfs://hadoop004:9000/data/ruozeinput.txt")
a: org.apache.spark.rdd.RDD[String] = hdfs://hadoop004:9000/data/ruozeinput.txt MapPartitionsRDD[3] at textFile at <console>:24scala> a.collect
res1: Array[String] = Array(hello       hello   hello, world    world, john)    2、尝试读取文件夹：
scala> val b=sc.textFile("hdfs://hadoop004:9000/data/")
b: org.apache.spark.rdd.RDD[String] = hdfs://hadoop004:9000/data/ MapPartitionsRDD[5] at textFile at <console>:24scala> b.collect
res2: Array[String] = Array(hello       hello   hello, world    world, john, hello   hello    hello, world    world, john, hello      hello   hello, world    world, john)3、测试把这个文件进行压缩了再上传hdfs进行读取：
[hadoop@hadoop004 ~]$ zip -r data.zip data/adding: data/ (stored 0%)adding: data/ruozeinput.txt (deflated 100%)
[hadoop@hadoop004 ~]$ hdfs dfs -put data.zip /data/
20/06/06 04:33:09 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
[hadoop@hadoop004 ~]$ hdfs dfs -ls /datascala> val c = sc.textFile("hdfs://hadoop004:9000/data/data.zip")
c: org.apache.spark.rdd.RDD[String] = hdfs://hadoop004:9000/data/data.zip MapPartitionsRDD[7] at textFile at <console>:24scala> c.collect
res3: Array[String] = Array(PK??,

3、The textFile method also takes an optional second argument for controlling the number of partitions of the file. By default, Spark creates one partition for each block of the file (blocks being 128MB by default in HDFS), but you can also ask for a higher number of partitions by passing a larger value. Note that you cannot have fewer partitions than blocks.

• textFile能够控制分区数，默认的，Spark为每一个block上的文件创建了分区，不能设置比block小的文件；就是textFile minPartitions的这个参数，控制partition的数量

Apart from text files, Spark’s Scala API also supports several other data formats：

• WholeTextFile：（返回值是filename文件名称,content文件内容）

scala> val rdd=sc.wholeTextFiles("hdfs://hadoop004:9000/data/")
rdd: org.apache.spark.rdd.RDD[(String, String)] = hdfs://hadoop004:9000/data/ MapPartitionsRDD[3] at wholeTextFiles at <console>:24scala> rdd.collect
res1: Array[(String, String)] =
Array((hdfs://hadoop004:9000/data/ruozeinput.txt,"hello hello   hello
world   world
john
"), (hdfs://hadoop004:9000/data/ruozeinput2.txt,"hello  hello   hello
world   world
john
"), (hdfs://hadoop004:9000/data/ruozeinput3.txt,"hello  hello   hello
world   world
john
"))

对于sequenceFile：

• 测试使用Spark读取sequenceFile格式的文件

其它一些注意事项：

• 可以使用HadoopRDD得到一个RDD，测试如下：

1、在spark-shell中做测试：                                                                  
scala> val distFile=sc.textFile("hdfs://192.168.179.155:9000/data/input/ruozeinput.txt")
distFile: org.apache.spark.rdd.RDD[String] = hdfs://192.168.179.155:9000/data/input/ruozeinput.txt MapPartitionsRDD[13] at textFile at <console>:24scala> distFile.saveAsTextFile("hdfs://192.168.179.155:9000/data/output")2、在目录中读取：
[hadoop@hadoop004 data]$ hdfs dfs -ls /data/output
20/06/06 10:17:35 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Found 3 items
-rw-r--r--   3 hadoop supergroup          0 2020-06-06 10:17 /data/output/_SUCCESS
-rw-r--r--   3 hadoop supergroup         18 2020-06-06 10:17 /data/output/part-00000
-rw-r--r--   3 hadoop supergroup         17 2020-06-06 10:17 /data/output/part-00001
[hadoop@hadoop004 data]$ hdfs dfs -text /data/output/part-00000
20/06/06 10:17:51 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
hello   hello   hello
[hadoop@hadoop004 data]$ hdfs dfs -text /data/output/part-00001
20/06/06 10:18:19 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
world   world
john

出现的问题：

输出文件为什么是有2个，和什么有关？

• 调优点：如何设置分区数量，避免生成太多为空文件