什么是Spark？

关于Spark具体的定义，大家可以去阅读官网或者百度关于Spark的词条，在此不再赘述。从一个野生程序猿的角度去理解，作为大数据时代的一个准王者，Spark是一款主流的高性能分布式计算大数据框架之一，和MapReduce，Hive，Flink等其他大数据框架一起支撑了大数据处理方案的一片天空。笔者所在的公司，集群里面有数千台高配机器搭载了Spark(还有Hive和Flink)，用来处理千亿万亿级别的大数据。黑体字内容基本就是对Spark的一个概括。

什么是RDD？

套用一段关于RDD的常规解释，RDD 是 Spark 提供的最重要的抽象概念，它是一种有容错机制的特殊数据集合，可以分布在集群的结点上，以函数式操作集合的方式进行各种并行操作。通俗点来讲，可以将 RDD 理解为一个分布式对象集合，本质上是一个只读的分区记录集合。每个 RDD 可以分成多个分区，每个分区就是一个数据集片段。一个 RDD 的不同分区可以保存到集群中的不同结点上，从而可以在集群中的不同结点上进行并行计算。大家听懂了啵？

Again，用一个野生程序猿的话来说，RDD就是一个数据集，里面包含着我们要处理的千亿万亿数据，类似于Java里面的ArrayList，Python里面的list。不同的是，Spark基于RDD提供了一大堆很好用的函数(算子)，专门来处理大数据。

Next？

作为一个人狠话不多的野生程序猿，就喜欢生猛地直接上代码。No BB, show you the code.

Wait.

思维缜密的我，还是得BB一句，工欲善其事必先利其器。想玩起来Spark，请先做好一下准备，以下以Windows举例说明，Linux雷同。环境已经搭好的同学们，请忽略这一步，直接往下看。

#1，备好IDE

Java/Scala，请安装好宇宙 第二的IDE，IDEAL(全名 IntelliJ IDEA)，社区版即可，无需破解。Scala需要在IDEAL的Plugins里面，安装Scala插件。

Python，也请安装好世界第三的IDE，PyCharm，社区版即可，无需破解。

IDEA和PyCharm都出自于一个很厉害的软件公司，JetBrains，这家公司以一己之力，扛起了编程界的好几门主流语言的IDE。

#2，Spark

不管大家吃饭的家伙是Java，Scacla，还是Python，建议大家都去装一个Python，宇宙第二的编程语言(宇宙第一的语言是PHP)，太好用了。

---如果是Python，直接在命令行执行pip install pyspark，即可安装Spark。装好之后，Java/Scala也可以用来操作Spark。

---如果是Java/Scala，如果大家电脑上有安装Python，直接按照上一步操作装好pyspark之后，Java/Scala就可以共用。

如果老铁们不愿意安装Python，就需要自行去Spark官网下载相应版本，解压后，把spark的bin路径添加到Windows环境变量。(Windows下可能会报一个找不到null的错误，莫慌，需要自行下载Hadoop，以及对应版本的winutils，然后用winutils bin里面的内容新覆盖hadoop bin文件夹)

走到这一步，准备各做就绪。

祭出代码

Part I --- 测试数据

先准备点测试数据。数据包含2个字段，结构：name　　score,每列用\t分割。代码如下：

Python版本测试数据，name长度可以修改get_random_string参数，数据条数请根据自己电脑的配置修改loops参数。

import stringimport randomfile_data = 'seed'file_save = 'result'def get_random_string(size: int) -> str: stack = string.digits + string.ascii_letters rs = [stack[random.randrange(len(stack))] for _ in range(size)] # return ''.join(rs)def produce_seed(): loops = 100000 rs = ['{}\t{}'.format(get_random_string(4), random.randint(0, loops)) for _ in range(loops)] with open(file_data, 'w') as f: f.write('\n'.join(rs))if __name__ == '__main__': produce_seed()

Scala/Java版本测试数据，同样的，请老铁们自行修改name长度和数据总行数。

import java.io.Fileimport java.utilimport org.apache.commons.io.FileUtilsimport org.apache.commons.lang3.RandomStringUtilsimport scala.util.{Random}object Course { val dataFile = "seed" val savePath = "result" def main(args: Array[String]): Unit = { produceSeed(100000) } def produceSeed(loops:Int):Unit = { val file = new File(dataFile) val data = new util.ArrayList[String]() var str = "" (1 to loops).foreach(_ => { str = RandomStringUtils.randomAlphanumeric(10) data.add(s"$str\t${Random.nextInt(loops)}") if (data.size()>0 && data.size() % 10000==0) { FileUtils.writeLines(file,"UTF-8",data,true) data.clear() } }) if (data.size()>0) { FileUtils.writeLines(file,"UTF-8",data,true) } }}

附上pom，

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>groupId</groupId> <artifactId>scala3</artifactId> <version>1.0-SNAPSHOT</version> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <scala.version>2.11.12</scala.version> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <maven.compiler.compilerVersion>1.8</maven.compiler.compilerVersion> </properties> <dependencies> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.11</artifactId> <version>2.4.5</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>2.4.5</version> </dependency> </dependencies></project>

Part II -- 生成RDD

为了给大家尽可能多展示一些算子，下面的示例部分算子可能有些冗余，大家可以根据需求自行修改。

#Python版本：

需求：从源数据中找出第二列大于等于500的数据，并保存

#encoding=utf-8import shutilfrom pyspark.sql import SparkSessionfile_data = 'seed'file_save = 'result'def remove(filename: str): try: shutil.rmtree(filename) except: passdef rdd_sample_1(): ''' 选出第二列 >= 500的数据，并输出file_save ''' remove(file_save) def map1(row): parts = row.split('\t') return parts[0], int(parts[1]) # 如果数据很大，可以在textFile之后，用repartition进行重分区 rdd = sc.textFile(file_data) rdd1 = rdd \ .map(map1) \ .filter(lambda r: r[1] >= 500) \ .map(lambda r: '{}\t{}'.format(r[0], r[1])) \ .coalesce(1) # 或者直接在map partition阶段就进行挑选，效率要高一些。但是需要调大对应的内存，否则容易造成内存溢出 def map2(iter): for row in iter: try: parts = row.split('\t') if int(parts[1]) >= 500: yield row except Exception as e: print(e) # rdd2 = rdd.mapPartitions(map2).coalesce(1) # 大家2种方式选其一即可。这里选择第1种 rdd1.saveAsTextFile(file_save)if __name__ == '__main__': spark = SparkSession.builder.appName('pyspark').master('local[*]').getOrCreate() sc = spark.sparkContext sc.setLogLevel("ERROR") # rdd_sample_1() # spark.stop()

Scala版本：

需求同Python版本

import java.io.Fileimport java.utilimport org.apache.commons.io.FileUtilsimport org.apache.commons.lang3.RandomStringUtilsimport org.apache.spark.sql.SparkSessionimport scala.collection.mutable.ListBufferimport scala.util.{Random, Try}object Course { val dataFile = "seed" val savePath = "result" val spark = SparkSession .builder() .appName("scala-spark") .master("local") .config("spark.sql.shuffle.partitions", "1000") .config("mapreduce.job.reduces",5) .getOrCreate() val sc = spark.sparkContext def main(args: Array[String]): Unit = { rddSample1() } def rddSample1(): Unit = { delete(savePath) // val rdd = sc.textFile(dataFile) // 第一种方式 rdd.filter(_.split("\t")(1).toInt >= 500) //.saveAsTextFile(savePath) // 第二种方式 rdd.map(v => { val parts = v.split("\t") if (parts(1).toInt >= 500) { v } else {""} }) .filter(!_.isEmpty)//.saveAsTextFile(savePath) // 第三种方式，如果你机器或者内存够大，可以用以下方式，效率更高 rdd.mapPartitions(iterator => { val rs = ListBuffer[String]() var parts = Array[String]() iterator.foreach(v => { parts = v.split("\t") if (parts(1).toInt >= 500) rs += v }) // rs.iterator }) .saveAsTextFile(savePath) // 以上3种方式任选其一，最后用saveAsTextFile保存即可 } def produceSeed(loops:Int):Unit = { val file = new File(dataFile) val data = new util.ArrayList[String]() var str = "" (1 to loops).foreach(_ => { str = RandomStringUtils.randomAlphanumeric(10) data.add(s"$str\t${Random.nextInt(loops)}") if (data.size()>0 && data.size() % 10000==0) { FileUtils.writeLines(file,"UTF-8",data,true) data.clear() } }) if (data.size()>0) { FileUtils.writeLines(file,"UTF-8",data,true) } } def delete(path:String):Try[Unit] = { Try(FileUtils.deleteDirectory(new File(path))) }}