Spark简介课件.ppt

Actions .foreach(println) 风格：把函数println当作参数传递给函数foreach 例子 - 计算bad的个数 errorsRDD = inputRDD.filter(line.contains(error)) warningsRDD = inputRDD.filter(line.contains(warning)) badLinesRDD = errorsRDD.union(warningsRDD) println(badLinesRDD.count() ) badLinesRDD.take(1).foreach(println)//使用take()取前1个数据 Actions top() 排序(根据RDD中数据的比较器) takeSample(withReplacement, num, seed) 取样例，是否需要替换值。 countByValue() 返回一个map，表示唯一元素出现的个数 Spark的核心概念 向Spark传递函数： 需要注意的地方： 如果你传递的函数是一个对象的成员，或者包含一个对象中字段的引用(例如self.field)，Spark会把整个对象都发送到工作节点上，这样会比仅仅发送你关心的信息要大很多，而且有时候会带来一些奇怪的问题。 传送信息太多解决方法：我们可以把关心的字段抽取出来，只传递关心的字段。 奇怪问题的避免：序列化包含函数的对象，函数和函数中引用的数据都需要序列化(实现Java的Serializable interface)。 如果Scala中出现NotSerializableException，一般情况下，都是因为没序列化。 RDDs介绍 RDDs介绍 RDDs的创建方法 Scala的基础知识 RDDs介绍 RDDs Resilient distributed datasets(弹性分布式数据集，简写RDDs)。 一个RDD就是一个不可改变的分布式集合对象，内部由许多partitions(分片)组成，每个partition都包括一部分数据，这些partitions可以在集群的不同节点上计算 Partitions是Spark中的并行处理的单元。Spark顺序的，并行的处理partitions。 RDDs 是 Spark的分发数据和计算的基础抽象类，是Spark的核心概念。 RDD可以包含 Python, Java, 或者 Scala中的任何数据类型，包括用户自定义的类。 在Spark中，所有的计算都是通过RDDs的创建，转换，操作完成的。 RDD具有lineage graph(血统关系图)。 RDDs的创建方法 Driver program中创建RDDs： 把一个存在的集合传给SparkContext’s parallelize()方法。这种方法，一般只适用于学习时。 例子： val lines = sc.parallelize(List(spark, bigdatastudy)) val rdd = sc.parallelize(Array(1, 2, 2, 4), 4) ... 注意一下RDD的类型 第一个参数是：待并行化处理的集合 第二个参数是：分区个数 RDDs的创建方法 加载外部数据集: 例子：使用textFile() 加载 val rddText= sc.textFile(../../testfile/helloSpark) val rddHdfs = sc.textFile(hdfs:///some/path.txt) Scala的基础知识 Scala的变量声明 在Scala中创建变量的时候，必须使用val或者var Val，变量值不可修改，一旦分配不能重新指向别的值 Var，分配后，可以指向类型相同的值。 Scala的基础知识 Scala的变量声明 val lines= sc.textFile(../../testfile/helloSpark) lines= sc.textFile(../../testfile/helloSpark2) ... console: error: reassignment to val var lines2= sc.textFile(../../testfile/helloSpark) lines2= sc.textFile(../../testfile/helloSpark2) 可以重新声明变量 val lines= sc.textFile(../../testfile/helloSpark2) Scala的基础知识 Scala的匿名函数 像Python的lambda 函数 lines.filter(line = line.contains(world)) ... 我们定义一个匿名函数，接收一个参