Spark机器学习技术及应用 课件 第1-6章 大数据与Spark-线性回归及应用

大数据与Spark第1章本章主要知识点：大数据概念及来源大数据处理方式介绍Spark概述Spark3.5核心—ML1什么是大数据我们先来看什么是大数据？想了解这个之前你必须知道什么是数据，其实文本，声音，图片，视频都是数据。想想你用手机数据线连上电脑的时候都传了些什么吧，那都是各种形式的数据。那么大数据如何定义呢？大数据指的就是数据体量达到了一定的级别，而我们现有的算法和工具无法在合理的时间内给予处理。才可以称为大数据。当然大数据还包括多样性（Variety），价值密度低（Valueless），处理速度快（Velocity）等特点。但最重要特点还是数据量（Volume）要大。我们知道描述一个东西很大的时候是需要带上单位的。比方说，姚明很高，身高230。这样描述显然不准确，是cm还是mm，如果230毫米那它是个模型。同样道理大数据也需要带有度量单位，下面是一些数据单位之间的换算关系。1什么是大数据1什么是大数据其实这些单位越来越大，是为大数据而生的，本来没有这些单位的。1PB就相当于美国国家图书馆藏书的所有内容的和。而google每天都在处理20pb的数据。一般认为达到PB级别的数据才可以称为大数据。里面最大的单位YB，有家统计机构给出1YB相当于世界上所有海滩上的沙子总和，准不准确无法说，反正谁都没空去数一数，你说相当于地球上所有空气的总和好像也没有什么不行。只是说明数据体量到了一个海量的级别。当然还有更大的单位没有列出，比如比YB更大还有NB。数据增长不停止的话，单位定义不会停止。2大数据的来源和数据分析的关键要素大数据的来源非常广泛,如信息管理系统、网络信息系统、、物联网系统、科学实验系统等,其数据类型包括结构化数据、半结构化数据和非结构化数据。主要有三个来源，一是传统互联网企业依旧在产生巨大的数据，比如京东，淘宝等等。比如淘宝双11的当天，交易额可以突破千亿，由交易产生的数据高达46g，这仅仅是一天的数据，注意这是顾客购物记录的文本数据，不是46集电视连续剧。第二个来源是物联网的发展带来了大数据。应该是我们已经进入了物联网时代，也即是在原来只有计算机组成的互联网基础上，加入许多非计算机节点。大街小巷的监控每天都在记录视频数据，物流中转站每天都在用手持设备扫描货物入库出库，还有门禁，大家刷的校园卡消费数据，另外还有家居智能产品等等，这些物联网设备在城市的每个角落随处可见，所以现在也有了智慧城市，智慧地球的概念，这是大数据很重要的一个来源。2大数据的来源和数据分析的关键要素第三个大家最熟悉就是移动应用快速发展，即移动通信数据。大家都成了低头族，聊着qq，发着微信，顾不上跟人打招呼，都在忙着造大数据。移动应用也是大数据的重要来源。麦肯锡全球研究所给出的定义是：一种规模大到在获取、存储、管理、分析方面大大超出了传统数据库软件工具能力范围的数据集合，具有海量的数据规模、快速的数据流转、多样的数据类型和价值密度低四大特征。大数据分析需要具备哪些知识呢?2大数据的来源和数据分析的关键要素数据挖掘算法大数据分析的理论核心就是数据挖掘算法，各种数据挖掘的算法基于不同的数据类型和格式才能更加科学的呈现出数据本身具备的特点，也正是因为这些被全世界统计学家所公认的各种统计方法(可以称之为真理)才能深入数据内部，挖掘出公认的价值。另外一个方面也是因为有这些数据挖掘的算法才能更快速的处理大数据，如果一个算法得花上好几年才能得出结论，那大数据的价值也就无从说起了。2大数据的来源和数据分析的关键要素预测性分析能力大数据分析最终要的应用领域之一就是预测性分析，从大数据中挖掘出特点，通过科学的建立模型，之后便可以通过模型带入新的数据，从而预测未来的数据。2大数据的来源和数据分析的关键要素语义引擎大数据分析广泛应用于网络数据挖掘，可从用户的搜索关键词、标签关键词、或其他输入语义，分析，判断用户需求，从而实现更好的用户体验和广告匹配。2大数据的来源和数据分析的关键要素数据质量和数据管理大数据分析离不开数据质量和数据管理，高质量的数据和有效的数据管理，无论是在学术研究还是在商业应用领域，都能够保证分析结果的真实和有价值。大数据分析的基础就是以上五个方面，当然更加深入大数据分析的话，还有很多很多更加有特点的、更加深入的、更加专业的大数据分析方法。3Spark概述ApacheSpark是专为大规模数据处理而设计的快速通用的计算引擎。Spark是UCBerkeleyAMPlab（加州大学伯克利分校的AMP实验室）所开源的类HadoopMapReduce的通用并行框架，Spark拥有HadoopMapReduce所具有的优点；但不同于MapReduce的是Job中间输出结果可以保存在内存中，从而不再需要读写HDFS，因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。Spark是一种与Hadoop相似的开源集群计算环境，但是两者之间还存在一些不同之处，这些有用的不同之处使Spark在某些工作负载方面表现得更加优越，换句话说，Spark启用了内存分布数据集，除了能够提供交互式查询外，它还可以优化迭代工作负载。3Spark概述Spark是在Scala语言中实现的，它将Scala用作其应用程序框架。与Hadoop不同，Spark和Scala能够紧密集成，其中的Scala可以像操作本地集合对象一样轻松地操作分布式数据集。尽管创建Spark是为了支持分布式数据集上的迭代作业，但是实际上它是对Hadoop的补充，可以在Hadoop文件系统中并行运行。通过名为Mesos的第三方集群框架可以支持此行为。Spark由加州大学伯克利分校AMP实验室（Algorithms,MachinesandPeopleLab）开发，可用来构建大型的、低延迟的数据分析应用程序。Spark是一个简单的大数据处理框架，它可以帮助程序设计人员和数据分析人员在不了解分布式底层细节的情况下，编写一个简单的数据处理程序就可以对大数据进行分析计算。3Spark概述Spark通过HDFS使用自带的和自定义的特定数据格式（RDD、DataFrame），基本上可以按照程序设计人员的要求处理任何数据（音乐、电影、文本文件、Log记录等），而不论数据类型是什么样的。编写相应的Spark处理程序，可以帮助用户获得任何想要的答案。有了Spark后，再没有数据被认为是过于庞大而不好处理或不好存储的，从而解决了之前无法解决的、对海量数据进行分析的问题，便于发现海量数据中潜在的价值。4Spark3.5机器学习库机器学习是一项强大的技术，用于开发个性化、推荐和预测洞察力，以便提供更多样化、更以用户为中心的数据产品和服务。许多机器学习算法在执行过程中涉及大量的迭代计算。Spark是用于迭代处理的高效内存计算系统，目前构建基于Spark的机器学习包的项目和产品在学术界和工业界得到广泛应用。首先谈一下Spark机器学习包新旧版本MLlib的区别。ML和MLlib都是Spark中的机器学习库，都能满足目前常用的机器学习功能需求。Spark官方推荐使用ML，因为它功能更全面、更灵活，未来会主要支持ML，MLlib很有可能会被废弃。ML主要操作的是DataFrame，而MLlib操作的是RDD，也就是说二者面向的数据集不一样。相比于MLlib在RDD提供的基本操作，ML在DataFrame上的抽象级别更高，数据和操作耦合度更低。4Spark3.5机器学习库ML中的操作可以使用Pipeline，跟Sklearn一样，可以把很多操作（算法、特征提取、特征转换）以管道的形式串起来，然后让数据在这个管道中流动。ML中无论是什么模型，都提供了统一的算法操作接口，比如模型训练都是fit。01如果将Spark比作一颗闪亮的星星，那么其中最明亮、最核心的部分就是ML。ML是一个构建在Spark上、专门针对大数据处理的并发式高性能机器学习库，其特点是采用较为先进的迭代式、内存存储的分析计算，使数据的计算处理速度大大高于普通的数据处理引擎。02ML机器学习库还在不停地更新中，Apache的相关研究人员仍在不停地为其中添加更多的机器学习算法。目前ML中已经有通用的学习算法和工具类，包括统计、分类、回归、聚类、降维等034Spark3.5机器学习库图1-2ML的算法和工具类对预处理后的数据进行分析，从而获得包含着数据内容的结果。ML作为Spark的核心处理引擎，在诞生之初就为处理大数据而采用了“分治式”的数据处理模式，将数据分散到各个节点中进行相应的处理。通过数据处理的“依赖”关系，使处理过程层层递进。这个过程可以依据要求具体编写，好处是避免了大数据处理框架所要求进行的大规模数据传输，从而节省了时间、提高了处理效率。4Spark3.5机器学习库ML借助函数式程序设计思想，让程序设计人员在编写程序的过程中只需要关注其数据，而不必考虑函数调用顺序，不用谨慎地设置外部状态。所有要做的就是传递代表了边际情况的参数。ML采用Scala语言编写。Scala语言是运行在JVM上的一种函数式编程语言，特点是可移植性强，最重要的特点是“一次编写，到处运行”。借助于RDD或DataFrame数据统一输入格式，让用户可以在不同的IDE上编写数据处理程序，通过本地化测试后可以在略微修改运行参数后直接在集群上运行。对结果的获取更为可视化和直观，不会因为运行系统底层的不同而造成结果的差异与改变。4Spark3.5机器学习库ML是Spark的核心内容，也是其中最闪耀的部分。对数据的分析和处理是Spark的精髓，也是挖掘大数据这座宝山的金锄头。Spark3.5中的数据集使用DataFrame格式，并且支持使用管道API进行运算。它对机器学习算法的API进行了标准化，以便更轻松地将多种算法组合到单个管道或工作流中。有了Pipeline（见图1-3）之后，ML更适合创建包含从数据清洗到特征工程再到模型训练等一系列工作。5Spark前景在大数据领域，只有深挖数据科学领域，走在学术前沿，才能在底层算法和模型方面走在前面，从而占据领先地位。Spark的这种学术基因，使得它从一开始就在大数据领域建立了一定优势。无论是性能，还是方案的统一性，对比传统的Hadoop，优势都非常明显。Spark提供的基于RDD的一体化解决方案，将MapReduce、Streaming、SQL、MachineLearning、GraphProcessing等模型统一到一个平台下，并以一致的API公开，并提供相同的部署方案，使得Spark的工程应用领域变得更加广泛。Spark的代码活跃度很高，从Spark的版本演化看，足以说明这个平台旺盛的生命力以及社区的活跃度。尤其在2013年来，Spark进入了一个高速发展期，代码库提交与社区活跃度都有显著增长。以活跃度论，Spark在所有Aparch基金会开源项目中，位居前列。相较于其他大数据平台或框架而言，Spark的代码库最为活跃。5Spark前景Spark一个新兴的、能够便捷和快速处理海量数据的计算框架，得到了越来越多从业者的关注与重视。使用其中的ML能够及时准确地分析海量数据，从而获得大数据中所包含的各种有用信息。例如，经常使用的聚类推荐、向感兴趣的顾客推荐相关商品和服务；或者为广告供应商提供具有针对性的广告服务，并且通过点击率的反馈获得统计信息，进而有效地帮助他们调整相应的广告投放能力。2015年6月15日，IBM宣布了一系列ApacheSpark开源软件相关的措施，旨在更好地存储、处理以及分析大量不同类型的数据。IBM将在旧金山开设一家Spark技术中心，这一举措将直接教会3500名研发人员使用Spark来工作，并间接影响超过一百万的数据科学家和工程师，让他们更加熟悉Spark。5Spark前景相对于IBM对Spark的大胆采纳，其他一些技术厂商对Spark则是持相当保留的态度。IBM近年来将战略重点转向数据领域，在大数据、物联网、软件定义存储及Watson系统等领域投入大量资金。IBM在Spark开源软件方面的举动，将会对许多以Spark为框架协议的初创公司带来利益，最重要的是会使业界对Spark开源软件的接受度和应用率增加。因为Spark开源软件不仅对初创公司有利，对于一些大的数据项目来说也是非常好的解决方案。Spark将是大数据分析和计算的未来，定将会成为应用最为广泛的计算架构。越来越多的公司和组织选择使用Spark，不仅体现出使用者对大数据技术和分析能力要求越来越高，也体现出Spark这一新兴的大数据技术对于未来的应用前景越来越好。1236小结Spark是未来大数据处理的最佳选择，其最核心、最重要的部分就是ML。掌握了使用ML对数据进行处理的技能，可以真正使大数据“为我所用”。谢谢Spark3.5安装和开发环境配置第二章本章将介绍Spark的单机版安装方法和开发环境配置。ML是Spark数据处理框架的一个主要组件，其运行必须有Spark的支持。本书以讲解和演示ML原理和示例为主，在安装上将详细介绍基于IntellijIDEA、在Windows10操作系统上的单机运行环境，这也是Spark机器学习和调试的最常见形式，以便更好地帮助读者学习和掌握Spark程序编写精髓。本章主要知识点：Spark运行模式环境搭建Spark单机版的安装与配置写出第一个Spark程序目录01Spark的运行模式概述02单机模式下Spark环境安装与配置03WordCount实例04小结Spark的运行模式概述Spark的运行模式概述目前ApacheSpark支持三种分布式部署方式，分别是standalone、sparkonmesos和sparkonYARN，其中，第一种类似于MapReduce1.0所采用的模式，内部实现了容错性和资源管理，后两种则是未来发展的趋势，部分容错性和资源管理交由统一的资源管理系统完成：让Spark运行在一个通用的资源管理系统之上，这样可以与其他计算框架，比如MapReduce，公用一个集群资源，最大的好处是降低运维成本和提高资源利用率（资源按需分配）。Spark的运行模式概述1.Standalone模式即独立模式，自带完整的服务，可单独部署到一个集群中，无需依赖任何其他资源管理系统。从一定程度上说，该模式是其他两种的基础。借鉴Spark开发模式，我们可以得到一种开发新型计算框架的一般思路：先设计出它的standalone模式，为了快速开发，起初不需要考虑服务（比如master/slave）的容错性，之后再开发相应的wrapper，将standalone模式下的服务原封不动的部署到资源管理系统yarn或者mesos上，由资源管理系统负责服务本身的容错。目前Spark在standalone模式下是没有任何单点故障问题的，这是借助zookeeper实现的，思想类似于Hbasemaster单点故障解决方案。Spark的运行模式概述2.SparkOnMesos模式这是很多公司采用的模式，官方推荐这种模式（当然，原因之一是血缘关系）。正是由于Spark开发之初就考虑到支持Mesos，因此，目前而言，Spark运行在Mesos上会比运行在YARN上更加灵活，更加自然。目前在SparkOnMesos环境中，用户可选择两种调度模式之一运行自己的应用程序（可参考AndrewXia的“MesosSchedulingModeonSpark”）：1)粗粒度模式（Coarse-grainedMode）：每个应用程序的运行环境由一个Dirver和若干个Executor组成，其中，每个Executor占用若干资源，内部可运行多个Task（对应多少个“slot”）。应用程序的各个任务正式运行之前，需要将运行环境中的资源全部申请好，且运行过程中要一直占用这些资源，即使不用，最后程序运行结束后，回收这些资源。举个例子，比如你提交应用程序时，指定使用5个executor运行你的应用程序，每个executor占用5GB内存和5个CPU，每个executor内部设置了5个slot，则Mesos需要先为executor分配资源并启动它们，之后开始调度任务。另外，在程序运行过程中，mesos的master和slave并不知道executor内部各个task的运行情况，executor直接将任务状态通过内部的通信机制汇报给Driver，从一定程度上可以认为，每个应用程序利用mesos搭建了一个虚拟集群自己使用。Spark的运行模式概述2)细粒度模式（Fine-grainedMode）：鉴于粗粒度模式会造成大量资源浪费，SparkOnMesos还提供了另外一种调度模式：细粒度模式，这种模式类似于现在的云计算，思想是按需分配。与粗粒度模式一样，应用程序启动时，先会启动executor，但每个executor占用资源仅仅是自己运行所需的资源，不需要考虑将来要运行的任务，之后，mesos会为每个executor动态分配资源，每分配一些，便可以运行一个新任务，单个Task运行完之后可以马上释放对应的资源。每个Task会汇报状态给Mesosslave和MesosMaster，便于更加细粒度管理和容错，这种调度模式类似于MapReduce调度模式，每个Task完全独立，优点是便于资源控制和隔离，但缺点也很明显，短作业运行延迟大。Spark的运行模式概述3.SparkOnYARN模式这是一种很有前景的部署模式。但限于YARN自身的发展，目前仅支持粗粒度模式（Coarse-grainedMode）。这是由于YARN上的Container资源是不可以动态伸缩的，一旦Container启动之后，可使用的资源不能再发生变化，不过这个已经在YARN计划中了。sparkonyarn的支持两种模式：1)yarn-cluster：适用于生产环境；2)yarn-client：适用于交互、调试，希望立即看到app的输出yarn-cluster和yarn-client的区别在于yarnappMaster，每个yarnapp实例有一个appMaster进程，是为app启动的第一个container；负责从ResourceManager请求资源，获取到资源后，告诉NodeManager为其启动container。yarn-cluster和yarn-client模式内部实现还是有很大的区别。如果你需要用于生产环境，那么请选择yarn-cluster；而如果你仅仅是Debug程序，可以选择yarn-client。Spark的运行模式概述这三种分布式部署方式各有利弊，通常需要根据实际情况决定采用哪种方案。进行方案选择时，往往要考虑公司的技术路线（采用Hadoop生态系统还是其他生态系统）、相关技术人才储备等。上面涉及到Spark的许多部署模式，究竟哪种模式好这个很难说，需要根据你的需求，如果你只是测试SparkApplication，你可以选择local模式。而如果你数据量不是很多，Standalone是个不错的选择。当你需要统一管理集群资源（Hadoop、Spark等），那么你可以选择Yarn或者mesos，但是这样维护成本就会变高。为了方便学习Spark机器学习程序的开发和调试，本书采用Spark单机模式运行程序，下面会详细介绍具体安装过程。单机模式下Spark环境安装与配置Java8安装Windows10系统是目前最常见的操作系统，本节将讲解如何在Windows10系统中下载、使用Spark单机模式。ML是Spark大数据处理框架中的一个重要组件，广泛应用于各类数据的分析和处理。Scala是一种基于JVM的函数式编程语言，而Spark是借助于JVM运行的一个数据处理框架，因此首选安装Java。Java8安装从Java地址下载安装Java安装程序，地址为/technetwork/java/javase/downloads/index.html。单击JavaDownLoad，进入下载页面。这里推荐读者全新安装时使用Java8，如图2-1所示。Java8安装单击JDKDownload按钮，之后按需求选择Java的版本号。为了统一安装，这里全部选择64位Java安装文件进行下载，如图2-2所示。Java8安装双击下载后的文件，在默认路径安装Java，如图2-3所示，然后静待安装结束即可。笔者采用的是1.8.151版本，学习时只要比此版本高即可。Java8安装安装结束后需要对环境变量进行配置，首先右击“我的电脑”|“属性”选项，在弹出的对话框中下方单击“高级系统设置”选项，然后选中“高级”标签，单击“环境变量”按钮，在当前用户名下新建JAVA_HOME安装路径，即前面JDK安装所在的路径，如图2-4所示。Java8安装安装结束后需要对环境变量进行配置，首先右击“我的电脑”|“属性”选项，在弹出的对话框中下方单击“高级系统设置”选项，然后选中“高级”标签，单击“环境变量”按钮，在当前用户名下新建JAVA_HOME安装路径，即前面JDK安装所在的路径，如图2-4所示。Java8安装对CLASSPATH进行配置。注意，在“变量值”（路径）文本框中一定要在开头加上“.;”（不包括引号），如图2-6所示。Java8安装单击Windows10开始菜单，在“附件”里面找到“运行”，输入“cmd”命令，如图2-7所示。Java8安装输入命令后打开控制台界面，在打开的界面中输入“java”，如图2-8所示。Java8安装行后出现如图2-9所示的界面，说明Java已经配置好，可以运行Java程序了。Scala安装Scala的安装比较容易，直接下载相应的编译软件，下载之后双击程序直接安装。Scala会在安装过程中自行设置。我们需要下载的版本是Scala2.12.10，下载地址为。打开Scala网站首页，如图2-10所示。Scala安装单击DOWNLOAD按钮，进入下载界面，单击如图2-11所示的圈住的链接。日期不同，在首页默认下载的Scala版本也不尽相同，这里选用的是2.12.10版本。单击图2-11中所示的ALLdownloads按钮进入版本选择页面，如图2-12所示。Scala安装单击图2-12中所示画横线的按钮进入Scala2.12.10版本的下载页面，选择Windows版本，如图2-13所示。等待程序下载完成后，双击进行程序安装。与Java安装时类似，安装结束后对环境变量进行配置，首先右击“我的电脑”|“属性”菜单，在弹出的对话框中单击“高级系统设置”选项，然后选中“高级”标签，单击“环境变量”按钮。在当前用户名下新建SCALA_HOME安装路径，即前面Scala安装所在的路径，如图2-14所示。Scala安装设置PATH变量：找到系统变量下的PATH项，单击“编辑”按钮。在“变量值”文本框的最前面添加“%SCALA_HOME%\bin;”，如图2-15所示。跟前面运行Java命令一样，还是通过在“运行”对话框输入“cmd”命令打开命令控制台。输入“scala”，显示如图2-16所示，即可认为Scala安装完毕。IntellijIDEA开发环境安装IntellijIDEA是常用的Java编译器，也可以用来作为Spark单机版的调试器。IntellijIDEA有社区免费版和付费版，这里使用免费版即可。IntellijIDEA的下载地址为/idea/download/，选择右侧的社区免费版下载即可，如图2-17所示。双击下载的IntellijIDEA安装包就会自动进行安装。这里基本没有什么需要特别注意的事项，如果在安装过程中碰到问题，可以自行搜索解决。Scala插件的安装Scala是一种把面向对象和函数式编程理念加入静态类型语言中的语言，可以把Scala应用在很大范围的编程任务上，无论是小脚本还是大系统都可以用Scala实现。Scala运行在标准的Java平台上（JVM），可以与所有的Java库实现无缝交互。Spark的ML库是基于Java平台的大数据处理框架，因此可以自由选择最方便的语言进行编译处理。Scala天生具有简洁性和性能上的优势，并且可以在JVM上直接使用，使其成为Spark官方推荐的首选程序语言。因此，笔者推荐使用Scala语言作为Spark机器学习的首选语言。IntellijIDEA本身并没有安装Scala编译插件，因此在使用IntellijIDEA编译Scala语言编写的Spark机器学习代码之前需要安装Scala编译插件，其安装步骤如下：Scala插件的安装在桌面上找到已安装的IntellijIDEA图标，双击打开后等待读取界面（见图2-18）结束。由于IntellijIDEA是首次使用，因此之后会进入创建工程选项界面，如图2-19所示。Scala插件的安装因为需要使用Scala语言编译程序，所以这里建议读者先选择新建工程，验证是否可以使用Scala创建工程，如图2-20所示。从图2-20可以看到，其中并没有可以建立Scala工程的选项。也就是说，如果需要使用Scala，IntellijIDEA需要进一步配置相应的开发组件。这里单击FileSetting…菜单项打开Setting窗口，在左边单击Plugins选择插件，会出现如图2-21所示的界面（显示当前可以安装的插件）。Scala插件的安装显示的插件过多时，可以在Search文本框中输入“scala”搜索相应的Scala插件，如图2-22所示。找到Scala插件后，单击右侧的installplugin绿色按钮，等待一段时间，即可完成安装。如图2-23所示是安装好了的界面，Installed按钮是灰色的。Scala插件的安装安装完毕后，在NewProject选项下有一项新的项目“Scala”，如图2-24所示。单击项目，可以创建相关程序。至此，IntellijIDEA的Scala插件安装完毕。编写Java程序单击桌面上的IntellijIDEA标记，打开IntellijIDEA软件。这里建议读者先新建工程，单击新建工程对应项后，操作界面如图2-25所示。上面已经成功安装了Java、Scala以及通用编译器IntellijIDEA，下面带领读者正式使用IntellijIDEA创建Java与Scala的HelloWorld小程序。编写Java程序这里首先创建的是Java程序，因此在图2-25所示的对话框左侧列表中选择Java选项、右侧列表中勾选Kotlin/JVM复选框。单击ProjectSDK下拉列表，在弹出的菜单项中选择JDK…，按操作提示选择JavaJDK安装目录，结果如图2-26所示。IDE已经自动认出Java的版本号（见图2-27），此时可以使用IntellijIDEA创建一个Java程序。编写Java程序单击Next按钮后，给创建的文件起一个名字（见图2-28），然后单击Finish按钮，即可创建程序文件。在IDEA界面左侧，右击项目名下的src目录，弹出快捷菜单，单击New|JavaClass菜单项，如图2-29所示。在打开的对话框中填写名称HelloJava，如图2-30所示，单击OK按钮后创建一个新的Java程序。编写Java程序在IDEA弹出的界面右侧补充代码，如程序2-1所示。右击文件，在弹出的菜单中选择“Run'HelloJava.main()'”运行此程序，结果如图2-31所示。这里使用Java语言创建了一个新的JavaClass文件，用于对程序进行编写与编译。虽然在后续的学习中，Java语言并不是作为本书Spark的主要程序设计语言，但是对于Spark来说，Java语言仍旧是一个非常重要的语言基础，有无可替代的作用。publicclassHelloJava{publicstaticvoidmain(String[]args){System.out.print("helloJava");}}编写Scala程序单击IDE主界面上的File标签，新建一个工程，如图2-32所示。本节将继续使用IntellijIDEA编译器编译Scala程序，这是本书的一个重要的基础内容。进入工作界面，这里有两种可以新建Scala程序的方式，推荐使用第二种方式，即使用图2-33所示的左边框中的Scala选项和右边的IDEA程序。编写Scala程序单击Finish按钮后，进入存放位置设置和Scala编译器设置的页面（见图2-34），这里选择输入2.1.2节中安装的Scala目录地址。这里已经在2.1.2节中安装过Scala，因此直接选择查找已安装的SDK即可。有需要的话也可以直接单击Download…按钮，下载不同版本的Scala语言。新建项目，如图2-35所示。编写Scala程序单击Finish按钮后，进入存放位置设置和Scala编译器设置的页面（见图2-34），这里选择输入2.1.2节中安装的Scala目录地址。这里已经在2.1.2节中安装过Scala，因此直接选择查找已安装的SDK即可。有需要的话也可以直接单击Download…按钮，下载不同版本的Scala语言。新建项目，如图2-35所示。这里使用了两个编译器，分别是Java和Scala的SDK。对于Scala来说，其实质也是运行在Java虚拟机上的一种编译语言，需要获得JDK的支持。编写Scala程序右击左侧列表中的src列表项新建文件，如图2-36所示。需要注意的是，这里要新建一个ScalaClass文件。弹出的对话框中输入Scala文件名，单击OK按钮即可创建一个空的Scala程序。需要注意的是，类型必须为Object而非Class，如图2-37所示。这一点和Java程序不同。ObjecthelloScala{defmain(args:Array[String]):Unit={print("helloScala")}}编写Scala程序与Java编译时类似，右击文件名helloScala，在弹出的快捷菜单中选择“Run'helloScala'”命令，如图2-38所示。最终运行结果如图2-39所示Spark3.5单机版安装Spark单机版安装首先需要下载Spark预编译版本，网站地址为/。单击网页左边的

标签进入下载页面，如图2-40所示。本小节通过Windows10系统模拟了一个Spark运行环境，从而使得读者学习Spark机器学习更加方便、简单。Spark3.5单机版安装选择Spark的下载版本，因为笔者将在Windows10上虚拟出一个Spark的运行环境，因此建议读者下载安装Spark3.5.1的预编译版本。这里选用的是Spark3.5版本的文件，所以推荐读者也使用Spark3.5版本，如图2-41所示。下载后的文件是tgz格式的压缩文件。可能有读者使用Linux初步学习过Spark，但是单机版本的Spark与Linux不同，此时下载的tgz文件不要安装，直接使用Winrar软件或者Bandizip软件解压打开即可。在压缩包文件中，所有的jars包是Spark的核心文件，也是其运行和计算的主体，如图2-42所示。Spark3.5单机版安装要在IntellijIDEA上运行Spark项目，就必须把里面的jars包都加入ProjectStructure的Libraries中。单击IntellijIDEA菜单栏上的File选项，选择ProjectStructure，在弹出的对话框中单击左侧的Libraries选项，之后单击中部上方的+按钮，选择Java文件，添加刚才下载的jars包文件，如图2-43所示。添加后的lib文件库，如图2-44所示。Spark3.5单机版安装返回主界面，打开左边工程栏下的工程扩展文件库，也可以看到Spark核心文件已经被安装，如图2-45所示。基于Python语言的环境配置Spark3.5实现wordCount经典的wordCount（统计文章中的词频）是MapReduce入门必看的例子，可以称为分布式框架的HelloWorld，也是大数据处理人员必须掌握的入门技能：考察对基本Spark语法的运用，还是一个简单的自然语言处理程序。源代码参照Spark官网的wordCount中的示例代码，实现了text文件的单词计数功能，单词之间按照空格来区分，形式如图2-46所示。这是需要计数的数据内容，我们需要计算出文章中每个单词出现的次数，Spark代码如程序2-3所示。valspark=SparkSession

.builder .master("local") .appName("wordCount") .getOrCreate()valdata=spark.read.text("wc.txt")implicitvalencoder=org.apache.spark.sql.Encoders.STRINGdata.as[String].rdd.flatMap(_.split("")).map((_,1)).reduceByKey(_+_).collect().foreach(println)Spark3.5实现wordCount下面是对程序进行分析。（1）首先创建一个SparkSession()，目的是创建一个会话变量实例，告诉系统开始Spark计算。之后的master("local")启动本地化运算、appName("wordCount")设置本程序名称。（2）getOrCreate()的作用是创建环境变量实例，准备开始任务。（3）spark.read.text("c://wc.txt")的作用是读取文件。顺便提一下，此时的文件读取是按照正常顺序读取，本书后面章节会介绍如何读取特定格式的文件。这种形式读出来的格式为SparkDataFrame，并非之前的RDD形式。（4）flatMap()是Scala中提取相关数据按行处理的一个方法。在_.split("")中，_是一个占位符，代表传送进来的任意一个数据，对其按""分割。map((_,1))对每个字符进行计数，在这个过程中并不涉及合并和计算，只是单纯地将每个数据行中的单词加1。最后的reduceByKey方法对传递进来的数据按key值相加，最终形成wordCount计算结果。valspark=SparkSession

.builder .master("local") .appName("wordCount") .getOrCreate()valdata=spark.read.text("wc.txt")implicitvalencoder=org.apache.spark.sql.Encoders.STRINGdata.as[String].rdd.flatMap(_.split("")).map((_,1)).reduceByKey(_+_).collect().foreach(println)Spark3.5实现wordCount目前程序流程如图2-47所示。Spark3.5实现wordCount（5）collect()对程序进行启动。因为Spark编程的优化，很多方法在计算过程中属于lazy模式，操作是延迟计算的，需要等到有Action操作的时候，才会真正触发运算，所以需要一个显性启动支持。foreach(println)是打印的一个调用方法，打印出数据内容。具体打印结果如下：(cool,2)(spark,3)(hadoop,1)(bad,2)(good,2)(mllib,2)valspark=SparkSession

.builder .master("local") .appName("wordCount") .getOrCreate()valdata=spark.read.text("wc.txt")implicitvalencoder=org.apache.spark.sql.Encoders.STRINGdata.as[String].rdd.flatMap(_.split("")).map((_,1)).reduceByKey(_+_).collect().foreach(println)Spark3.5实现wordCount首先是数据的准备工作。这里为了简化起见，采用小数据集（本书将以小数据为主，演示Spark机器学习的使用和原理）。在C盘创建名为wc.txt的文本文件（数据位置：//DATA//D02//wc.txt），文件名也可以自行设置，内容如下：goodbadcoolhadoopsparkmllibgoodsparkmllibcoolsparkbadSpark3.5实现wordCountMapReduce实现wordCount可以与Spark对比的是MapReduce中wordCount程序的设计，如程序2-4所示。只是为了做对比，如果有同学想深入学习MapReduce程序设计，请参考相关的专业图书。小结小结IntellijIDEA是目前常用的Java和Scala程序设计以及框架处理软件，拥有较好的自动架构、辅助编码和智能控制等功能，有取代Eclipse的趋势。在Windows上对Spark进行操作解决了大部分学习人员欠缺大数据运行环境的烦恼，便于操作和研究基本算法，这对真实使用大数据集群进行数据处理有很大的帮助。在后面的章节中，笔者将着重介绍基于Windows单机环境下Spark的数据处理方法。这种在单机环境下相应程序的编写与集群环境下运行时的程序编写基本相同，部分程序稍作修改即可运行在集群中。本章介绍了如何安装和上手运行一个Spark3.5的程序，下一章将详解Spark3.5ML包的主要使用格式DataFrame。第1章Scala编程基础1基础语法如果你之前学过Java语言，并了解Java语言的基础知识，那么你能很快学会Scala的基础语法。Scala与Java之间有些小的区别，比如Scala语句末尾的分号;是可选的。我们可以认为Scala程序是对象的集合，通过调用彼此的方法来实现消息传递。下面会详细介绍Scala编程语言的基础语法和编程常识。1基础语法注释1.//单行注释开始于两个斜杠/*开头*/结尾2.多行注释，如你之前所见，看起来像这样1基础语法单行注释//打印并强制换行println("Helloworld!")println(10)//没有强制换行的打印print("Helloworld")1基础语法变量1//通过var或者val来声明变量2//val声明是不可变的，var声明是可修改的。不可变性是好事。3valx=10//x现在是104x=20//错误:对val声明的变量重新赋值5vary=106y=20//y现在是201基础语法数据类型Scala与Java有着相同的数据类型，下面列出了Scala支持的数据类型如，图2-1所示。图2-1Scala支持的数据类型Scala数据类型设置样例：valz:Int=101基础语法数据类型vala:Double=1.0注意从Int到Double的自动转型，结果是10.0,不是10：valb:Double=10.0布尔值：trueFalse布尔操作：!true//false!false//truetrue==false//false10>5//true1基础语法运算符数学运算：1+1//22-1//15*3//156/2//36/4//16.0/4//1.51基础语法字符串"Scalastringsaresurroundedbydoublequotes"'a'//Scala的字符。'不存在单引号字符串'<=这会导致错误String有常见的Java字符串方法："helloworld".length"helloworld".substring(2,6)"helloworld".replace("C","3")也有一些额外的Scala方法："helloworld".take(5)1基础语法字符串"helloworld".drop(5)字符串改写，留意前缀"s"：valn=45s"Wehave$napples"//=>"Wehave45apples"在要改写的字符串中使用表达式也是可以的：vala=Array(11,9,6)s"Myseconddaughteris${a(0)-a(2)}yearsold."//=>"Mysecond1基础语法字符串s"Wehavedoubletheamountof${n/2.0}inapples."//=>"Wehavs"Powerof2:${math.pow(2,2)}"//=>"Powerof2:4"添加"f"前缀对要改写的字符串进行格式化：f"Powerof5:${math.pow(5,2)}%1.0f"//"Powerof5:25"f"Squarerootof122:${math.sqrt(122)}%1.4f"//"Squarerootof12未处理的字符串，忽略特殊字符：1基础语法字符串raw"Newlinefeed:\n.Carriagereturn:\r."//=>"Newlinefeed:一些字符需要转义，比如字符串中的双引号："Theystoodoutsidethe\"RoseandCrown\""//=>"Theystoodoutsi三个双引号可以使字符串跨越多行，并包含引号：valhtml="""<formid="daform"><p>Pressbelo',Joe</p><inputtype="submit"></form>"""2函数Scala有函数和方法，二者在语义上的区别很小。Scala方法是类的一部分，而函数是一个对象，可以赋值给一个变量。换句话来说在类中定义的函数即是方法。2函数函数声明Scala函数声明格式如下：deffunctionName([参数列表]):[returntype]如果你不写等于号和方法主体，那么方法会被隐式声明为“抽象（abstract）”，于是包含它的类型也是一个抽象类型。2函数函数定义函数定义由一个def关键字开始，紧接着是可选的参数列表，一个冒号“：”和函数的返回类型，一个等于号“=”，最后是函数的主体。Scala函数定义格式如下：deffunctionName([参数列表]):[returntype]={functionbodyreturn[expr]}以上代码中returntype可以是任意合法的Scala数据类型。参数列表中的参数可以使用逗号分隔。2函数函数定义以下函数的功能是将两个传入的参数相加并求和：objectadd{defaddInt(a:Int,b:Int):Int={varsum:Int=0sum=a+breturnsum}}2函数函数调用Scala提供了多种不同的函数调用方式。以下是调用方法的标准格式：functionName(参数列表)如果函数使用了实例的对象来调用，我们可以使用类似Java的格式（使用.号）：[instance.]functionName(参数列表)实例如下：程序10-1TestFunc.scala01objectTestFunc{02defmain(args:Array[String]){03println("ReturnedValue:"+addInt(5,7));04}05defaddInt(a:Int,b:Int):Int={06varsum:Int=007sum=a+b0809returnsum10}11}3控制语句控制语句变量使用Scala对点和括号的要求非常宽松，注意其规则是不同的。这有助于写出读起来像英语的DSL（领域特定语言）和API（应用编程接口）。1to5valr=1to5r.foreach(println)rforeachprintln执行以上代码，输出结果为：1,2,3,4,5,12345

(5to1by-1)foreach(println)执行以上代码，输出结果为：5,4,3,2,1,3控制语句while循环运行一系列语句，如果条件为true，会重复运行，直到条件变为false。程序10-2TestWhile.scalavari=0while(i<10){println("i"+i);i+=1}3.dowhile循环类似while语句，区别在于判断循环条件之前，先执行一次循环的代码块。程序10-3TestDoWhile.scalavarx=0;do{println(x+"isstilllessthan10");3控制语句while循环x+=1}while(x<10)执行以上代码，输出结果。4.for循环for循环允许编写一个执行指定次数的循环控制结构。测试代码如10-4所示：程序10-4TestFor.scaladefmain(args:Array[String]){vara=0;//for循环for(a<-1to10){println("Valueofa:"+a);}}3控制语句条件语句ScalaIF...ELSE语句是通过一条或多条语句的执行结果（True或者False）来决定执行的代码块。valx=10if(x==1)println("yeah")if(x==10)println("yeah")if(x==11)println("yeah")if(x==11)println("yeah")elseprintln("nay")println(if(x==10)"yeah"else"nope")yeah6.break语句Scala语言中默认是没有break语句，但是在Scala2.8版本后可以使用另外一种方式来实现break语句。当在循环中使用break语句，在执行到该语句时，就会中断循环并执行循环体之后的代码块。Scala中break的语法有点不大一样，格式如下：//导入以下包importscala.util.control._//创建Breaks对象valloop=newBreaks;//在breakable中循环loop.breakable{//循环for(...){....//循环中断loop.break;}}4函数式编程Array（数组）Scala数组声明的语法格式：varz:Array[String]=newArray[String](3)或varz=newArray[String](3)数组的元素类型和数组的大小都是确定的，所以当处理数组元素时，我们通常使用基本的for循环。以下实例演示了数组的创建、初始化等处理过程：程序10-6TestArray1.scala执行以上代码，输出结果为：4函数式编程List（列表）List的特征是其元素以线性方式存储，集合中可以存放重复对象。以下列出了多种类型的列表：//字符串列表valsite:List[String]=List("mrchi的博客","Google","Baidu")//整型列表valnums:List[Int]=List(1,2,3,4)//空列表valempty:List[Nothing]=List()//二维列表valdim:List[List[Int]]=List(List(1,0,0),List(0,1,0),List(0,0,1))对于Scala列表的任何操作都可以使用这三个基本操作来表达，实例见程序10-7TestList.scala4函数式编程Set（集合）Set是最简单的一种集合。集合中的对象不按特定的方式排序，并且没有重复对象。对于Scala集合的任何操作都可以使用这三个基本操作来表达，实例如下:程序10-8TestSet.scala01objectTestSet{02defmain(args:Array[String]){03valsite=Set("mrchi的博客","Google","Baidu")04valnums:Set[Int]=Set()06println("第一网站是:"+site.head)07println("最后一个网站是:"+site.tail)08println("查看列表site是否为空:"+site.isEmpty)09println("查看nums是否为空:"+nums.isEmpty)10}11}4函数式编程Map（映射）Map是一种把键对象和值对象映射的集合，它的每一个元素都包含一对键对象和值对象。以下实例演示了以上三个方法的基本应用：程序10-9TestMap.scala01objectTestMap{02defmain(args:Array[String]){03valcolors=Map("red"->"#FF0000",04"azure"->"#F0FFFF",05"peru"->"#CD853F")07valnums:Map[Int,Int]=Map()09println("colors中的键为:"+colors.keys)10println("colors中的值为:"+colors.values)11println("检测colors是否为空:"+colors.isEmpty)12println("检测nums是否为空:"+nums.isEmpty)13}14}4函数式编程元组元组是不同类型的值的集合。与列表一样，元组也是不可变的，但与列表不同的是元组可以包含不同类型的元素。元组的值是通过将单个的值包含在圆括号中构成的。例如：valt=(1,3.14,"Fred")以上实例在元组中定义了三个元素，对应的类型分别为[Int,Double,java.lang.String]。此外也可以使用以下方式来定义：valt=newTuple3(1,3.14,"Fred")可以使用t._1访问第一个元素，t._2访问第二个元素等，如下所示：程序10-10TestTuple.scala4函数式编程OptionOption[T]表示有可能包含值的容器，也可能不包含值。ScalaIterator（迭代器）不是一个容器，更确切地说是逐一访问容器内元素的方法。ScalaOption（选项）类型用来表示一个值是可选的（有值或无值)。Option[T]是一个类型为T的可选值的容器：如果值存在，Option[T]就是一个Some[T]；如果不存在，Option[T]就是对象None。5模式匹配Scala提供了强大的模式匹配机制，应用也非常广泛。一个模式匹配包含了一系列备选项，每个都开始于关键字case。每个备选项都包含了一个模式及一到多个表达式。箭头符号=>隔开了模式和表达式。以下是一个简单的整型值模式匹配实例：objectTest{defmain(args:Array[String]){println(matchTest(3))}defmatchTest(x:Int):String=xmatch{case1=>"one"case2=>"two"case_=>"many"}}5模式匹配match对应Java里的switch，但是写在选择器表达式之后。即：选择器match{备选项}。match表达式通过以代码编写的先后次序尝试每个模式来完成计算，只要发现有一个匹配的case，剩下的case不会继续匹配。接下来我们来看一个不同数据类型的模式匹配：5模式匹配执行以上代码，输出结果为：2manyonescala.Int实例中第1个case对应整型数值1，第2个case对应字符串值two，第3个case对应类型模式，用于判断传入的值是否为整型，相比使用isInstanceOf来判断类型，使用模式匹配更好。第4个case表示默认的全匹配备选项，即没有找到其他匹配时的匹配项，类似switch中的default。5模式匹配使用样例类使用了case关键字的类定义就是样例类（caseclasses），样例类是一种特殊的类，经过优化后用于模式匹配。以下是样例类的简单实例：程序10-12TestPattern1.scala6类和对象类的定义类是对象的抽象，而对象是类的具体实例。类是抽象的，不占用内存，而对象是具体的，占用存储空间。类是用于创建对象的蓝图，它是一个定义包括在特定类型的对象中的方法和变量的软件模板。我们可以使用new关键字来创建类的对象，实例如下：6类和对象类的定义Scala中的类不声明为public，一个Scala源文件中可以有多个类。以上实例的类定义了两个变量x和y，一个方法：move，方法没有返回值。Scala的类定义可以有参数，称为类参数，如上面的xc、yc，类参数在整个类中都可以访问。接着可以使用new来实例化类，并访问类中的方法和变量：6类和对象}执行以上代码，输出结果为：x的坐标点:20y的坐标点:306类和对象继承Scala继承一个基类跟Java很相似，但需要注意以下几点：（1）重写一个非抽象方法必须使用override修饰符。（2）只有主构造函数才可以往基类的构造函数里写参数。（3）在子类中重写超类的抽象方法时，不需要使用override关键字。接下来让我们来看个实例：6类和对象继承Scala使用extends关键字来继承一个类。实例中Location类继承了Point类。Point称为父类（基类），Location称为子类。overridevalxc为重写了父类的字段。继承会继承父类的所有属性和方法，Scala只允许继承一个父类。实例如下：6类和对象继承执行以上代码，输出结果为：x的坐标点:20y的坐标点:30z的坐标点:20Scala重写一个非抽象方法，必须用override修饰符。7Scala异常处理Scala的异常处理与其他语言比如Java类似。Scala的方法可以通过抛出异常方法的方式来终止相关代码的运行，不必通过返回值。7Scala异常处理抛出异常Scala抛出异常的方法和Java一样，使用throw方法。例如，抛出一个新的参数异常：thrownewIllegalArgumentException7Scala异常处理捕获异常异常捕捉的机制与其他语言的处理方法一样，如果有异常发生，catch子句是按次序捕捉的。因此，在catch子句中，越具体的异常越要靠前，越普遍的异常越靠后。如果抛出的异常不在catch子句中，该异常则无法处理，会被升级到调用者处。捕捉异常的catch子句，语法与其他语言中不太一样。在Scala里，借用了模式匹配的思想来做异常的匹配，因此，在catch的代码里，是一系列case字句，如下例所示：}执行以上代码，输出结果为：Missingfileexceptioncatch语句里的内容跟match里的case是完全一样的。由于异常捕捉是按次序，如果把最普遍的异常Throwable写在最前面，则在它后面的case都捕捉不到，因此需要将它写在最后面。}3.finally语句finally语句用于执行不管是正常处理还是有异常发生时都需要执行的步骤，实例如下：程序10-17TestFinally.scala05objectTestFinally{06defmain(args:Array[String]){07try{08valf=newFileReader("input.txt")09}catch{10caseex:FileNotFoundException=>{11println("Missingfileexception")12}13caseex:IOException=>{14println("IOException")15}16}finally{17println("Exitingfinally...")18}19}20}8Trait（特征）ScalaTrait（特征）相当于Java的接口，实际上它比接口的功能还要强大。与接口不同的是，它还可以定义属性和方法的实现。一般情况下，Scala的类只能够继承单一父类，但是如果是Trait的话就可以继承多个，从结果来看就是实现了多重继承。Trait定义的方式与类类似，但它使用的关键字是trait，如下所示：traitEqual{defisEqual(x:Any):BooleandefisNotEqual(x:Any):Boolean=!isEqual(x)以上Trait由两个方法组成：isEqual和isNotEqual。isEqual方法没有定义方法的实现，isNotEqual定义了方法的实现。子类继承特征可以实现未被实现的方法。所以其实ScalaTrait更像Java的抽象类。8Trait（特征）以下演示了特征的完整实例：01traitEqual{02defisEqual(x:Any):Boolean03defisNotEqual(x:Any):Boolean=!isEqual(x)04}0506classPoint(xc:Int,yc:Int)extendsEqual{07varx:Int=xc08vary:Int=yc09defisEqual(obj:Any)=10obj.isInstanceOf[Point]&&11obj.asInstanceOf[Point].x==x12}1314objectTest{15defmain(args:Array[String]){16valp1=newPoint(2,3)17valp2=newPoint(2,4)18valp3=newPoint(3,3)1920println(p1.isNotEqual(p2))21println(p1.isNotEqual(p3))22println(p1.isNotEqual(2))23}24}9Scala文件I/OIO