建手机网站公司,育婴网站模板,柳州做网站人员,华为建站文章目录 前言步骤一#xff1a;下载安装包Spark的目录和文件 步骤二#xff1a;使用Scala或PySpark Shell本地 shell 运行 步骤3:理解Spark应用中的概念Spark Application and SparkSessionSpark JobsSpark StagesSpark Tasks 转换、立即执行操作和延迟求值窄变换和宽变换 S… 文章目录 前言步骤一下载安装包Spark的目录和文件 步骤二使用Scala或PySpark Shell本地 shell 运行 步骤3:理解Spark应用中的概念Spark Application and SparkSessionSpark JobsSpark StagesSpark Tasks 转换、立即执行操作和延迟求值窄变换和宽变换 Spark UI单机的应用程序计算巧克力豆的数量单机编译 Scala 程序 总结 前言
本文将讲解 Spark 的部署并通过三个简单的步骤来编写一个独立应用程序。我们将使用本地模式其中所有的处理都是在Spark shell中的一台机器上完成的——这是学习框架的一种简单方法迭代执行的方式可以及时反馈直接结果。使用Spark shell可以在编写复杂的Spark应用程序之前使用小数据集对Spark进行操作验证但是对于想要获得分布式执行好处的大数据集或生产环境建议使用YARN或Kubernetes部署模式。虽然Spark shell只支持Scala、Python和R但你可以用任何支持的语言(包括Java)编写Spark应用程序并使用Spark SQL发出查询。
步骤一下载安装包
进入Spark下载页面在步骤2的下拉菜单中选择“Pre-built for Apache Hadoop 3.3”然后点击步骤3中的“download Spark”链接(如图2-1所示)。图2 - 1. Apache Spark下载页面这将下载压缩包spark-3.5.1-bin-hadoop3.tgz。它包含在笔记本电脑上以本地模式运行Spark所需的所有与hadoop相关的二进制文件。或者如果要将它安装在现有的HDFS或Hadoop安装上可以从下拉菜单中选择匹配的Hadoop版本。如果想要以源码编译的方式部署可以在官方文档中相关内容。自Apache Spark 2.2发布以来只关心在Python中学习Spark的开发人员可以选择从PyPI存储库安装PySpark。如果你只用Python编程你不需要安装运行Scala、Java或R所需的所有其他库; 要从PyPI安装PySpark只需运行pip install PySpark。可以通过pip install pyspark[SQL, ML, MLlib]安装SQL, ML和MLlib的一些额外依赖项(如果只想要SQL依赖项也可以通过pip install pyspark[SQL])。 NOTE 需要在机器上安装Java 8或更高版本并设置JAVA_HOME环境变量。有关如何下载和安装Java的说明请参阅文档。 如果想以解释性shell模式运行R则必须先安装R然后再运行sparkR。要使用R进行分布式计算还可以使用R社区创建的开源项目sparklyr。
Spark的目录和文件
本文中的所有命令和指令都是在 Unix 系统上运行的。下载完tarball后cd到下载目录使用tar -xf spark-3.5.1-bin-hadoop3.tgz解压tarball内容其中内容如下:
$ cd spark-3.0.0-preview2-bin-hadoop2.7
$ ls
LICENSE R RELEASE conf examples kubernetes python yarn
NOTICE README.md bin data jars licenses sbinREADME.md
这个文件包含了关于如何使用Spark shell、如何从源代码构建Spark、如何运行独立的Spark示例、如何阅读Spark文档和配置指南的链接以及如何为Spark做出贡献的新的详细说明.
bin
顾名思义该目录包含用于与Spark交互的大多数脚本包括Spark shell (Spark -sql、pyspark、Spark -shell和sparkR)。我们将在后面使用这个目录中的shell和可执行文件使用Spark -submit提交一个独立的Spark应用程序并编写一个脚本在Kubernetes支持下运行Spark时构建和推送Docker 镜像。
sbin
该目录中的大多数脚本都是用于管理的用于在集群的各种部署模式下启动和停止Spark组件。
kubernetes
自从Spark 2.4发布以来这个目录包含了用于在Kubernetes集群上为Spark发行版创建Docker镜像的Dockerfiles。它还包含一个文件提供如何在构建Docker映像之前构建Spark发行版的说明。
data
该目录中填充了*.txt文件这些文件作为Spark组件的输入:MLlib、Structured Streaming和GraphX。
examples
Spark提供了Java、Python、R和Scala的示例可以在学习该框架时用到它们。
步骤二使用Scala或PySpark Shell
如前所述Spark附带了四个广泛使用的解释器它们就像交互式“shell”一样支持临时数据分析:pyspark、Spark -shell、Spark-sql和sparkR。这些shell已经支持连接到集群并允许你将分布式数据加载到Spark worker的内存中。无论你是在处理千兆字节的数据还是小数据集Spark shell都有助于快速学习Spark。要启动PySpark, cd到 bin 目录并输入PySpark启动shell。如果你已经从PyPI安装了PySpark那么只需输入PySpark就足够了:
$ pyspark
Python 3.7.3 (default, Mar 27 2019, 09:23:15)
[Clang 10.0.1 (clang-1001.0.46.3)] on darwin
Type help, copyright, credits or license for more information.
20/02/16 19:28:48 WARN NativeCodeLoader: Unable to load native-hadoop library
for your platform... using builtin-java classes where applicable
Welcome to____ __/ __/__ ___ _____/ /___\ \/ _ \/ _ / __/ _//__ / .__/\_,_/_/ /_/\_\ version 3.0.0-preview2/_/Using Python version 3.7.3 (default, Mar 27 2019 09:23:15)
SparkSession available as spark.spark.version
3.0.0-preview2要用Scala启动一个类似的Spark shell, cd到bin目录并输入Spark -shell:
$ spark-shell
20/05/07 19:30:26 WARN NativeCodeLoader: Unable to load native-hadoop library
for your platform... using builtin-java classes where applicable
Spark context Web UI available at http://10.0.1.7:4040
Spark context available as sc (master local[*], app id local-1581910231902)
Spark session available as spark.
Welcome to____ __/ __/__ ___ _____/ /___\ \/ _ \/ _ / __/ _//___/ .__/\_,_/_/ /_/\_\ version 3.0.0-preview2/_/Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_241)
Type in expressions to have them evaluated.
Type :help for more information.
scala spark.version
res0: String 3.0.0-preview2
scala本地 shell 运行
Spark计算被表示为算子。然后这些算子被转换成低级的基于rdd的字节码作为任务分发给Spark的执行器执行。让我们看一个简短的示例其中我们以DataFrame的形式读取文本文件显示读取的字符串示例并计算文件中的总行数。这个简单的例子说明了高级结构化api的使用。DataFrame上的show(10, false)操作只显示前10行不截断。默认情况下截断布尔标志为true。下面是它在Scala shell中的样子:
scala val strings spark.read.text(../README.md)
strings: org.apache.spark.sql.DataFrame [value: string]scala strings.show(10, false)
------------------------------------------------------------------------------
|value |
------------------------------------------------------------------------------
|# Apache Spark |
| |
|Spark is a unified analytics engine for large-scale data processing. It |
|provides high-level APIs in Scala, Java, Python, and R, and an optimized |
|engine that supports general computation graphs for data analysis. It also |
|supports a rich set of higher-level tools including Spark SQL for SQL and |
|DataFrames, MLlib for machine learning, GraphX for graph processing, |
| and Structured Streaming for stream processing. |
| |
|https://spark.apache.org/ |
------------------------------------------------------------------------------
only showing top 10 rowsscala strings.count()
res2: Long 109
scala让我们看一个使用Python解释性shell pyspark的类似示例:
$ pyspark
Python 3.7.3 (default, Mar 27 2019, 09:23:15)
[Clang 10.0.1 (clang-1001.0.46.3)] on darwin
Type help, copyright, credits or license for more information.
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform
WARNING: Use --illegal-accesswarn to enable warnings of further illegal
reflective access operations
WARNING: All illegal access operations will be denied in a future release
20/01/10 11:28:29 WARN NativeCodeLoader: Unable to load native-hadoop library
for your platform... using builtin-java classes where applicable
Using Sparks default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to WARN.
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use
setLogLevel(newLevel).
Welcome to____ __/ __/__ ___ _____/ /___\ \/ _ \/ _ / __/ _//__ / .__/\_,_/_/ /_/\_\ version 3.0.0-preview2/_/Using Python version 3.7.3 (default, Mar 27 2019 09:23:15)
SparkSession available as spark.strings spark.read.text(../README.md)strings.show(10, truncateFalse)
------------------------------------------------------------------------------
|value |
------------------------------------------------------------------------------
|# Apache Spark |
| |
|Spark is a unified analytics engine for large-scale data processing. It |
|provides high-level APIs in Scala, Java, Python, and R, and an optimized |
|engine that supports general computation graphs for data analysis. It also |
|supports a rich set of higher-level tools including Spark SQL for SQL and |
|DataFrames, MLlib for machine learning, GraphX for graph processing, |
|and Structured Streaming for stream processing. |
| |
|https://spark.apache.org/ |
------------------------------------------------------------------------------
only showing top 10 rows strings.count()
109要退出任何Spark shell按Ctrl-D。这种与Spark shell的快速交互不仅有利于快速学习也有利于快速验证实验。我们使用高级结构化api将文本文件读入Spark DataFrame而不是RDD目前基本上已经很少直接使用 RDD 去操作数据而是使用 API。 NOTE 在高级结构化api中表达的每一个计算都被分解为低级的RDD操作然后转换为Scala字节码供执行器的jvm使用。这个生成的RDD操作代码对用户来说是不可访问的也与面向用户的RDD api不一样。 步骤3:理解Spark应用中的概念
要理解我们的示例代码在底层发生了什么需要熟悉Spark应用程序的一些关键概念以及代码如何作为任务在Spark执行器之间转换和执行:Application
使用Spark的 APIs 构建在Spark上的用户程序, 它由集群上的 driver 和 executors 组成。
SparkSession
它提供了与底层Spark功能交互的入口点的一个实例化对象并允许使用Spark的api对Spark进行编程。在交互式Spark shell中Spark driver 会自动实例化一个SparkSession而在Spark应用程序中我们自己可以创建一个SparkSession对象。
Job
由多个任务组成的并行计算这些任务在响应Spark操作(例如save() collect())时产生。
Stage
每个任务被分成更小的任务集称为阶段这些阶段相互依赖。
Task
将被发送到Spark executor 的操作命令或单个执行单元。
Spark Application and SparkSession
每个Spark application 的核心是Spark driver 程序它创建一个SparkSession对象。当你使用Spark shell时driver 是shell的一部分并且创建了SparkSession对象(可通过变量Spark访问)正如在启动shell时在前面的示例中看到的那样。在这些示例中因为在笔记本电脑上本地启动了Spark shell所以所有操作都在本地运行在单个JVM中运行。但是你可以像在本地模式下一样轻松地启动Spark shell来在集群上并行分析数据。命令Spark -shell——help或pyspark——help将向您展示如何连接到Spark集群管理器。图2-2显示了Spark在集群上的执行情况。图2 - 2. 在Spark的分布式架构中Spark组件通过Spark driver 进行通信一旦有了SparkSession就可以使用api对Spark进行编程来执行Spark操作。
Spark Jobs
在与Spark shell的交互会话中Driver 将我们的Spark应用程序转换为一个或多个Spark作业(图2-3)。然后将每个作业转换为DAG。本质上这就是Spark的执行计划其中DAG中的每个节点可以是单个或多个Spark阶段。图2 - 3. 创建一个或多个Spark作业的Spark Driver
Spark Stages
作为DAG节点的一部分阶段是根据可以串行或并行执行的算子创建的(图2-4)。并非所有的Spark 算子都可以在一个阶段中发生因此它们可以被划分为多个阶段。通常阶段是在运算符的计算边界上划分的在那里它们规定了Spark executor 之间的数据传输。图2 - 4. 创建一个或多个阶段的Spark job
Spark Tasks
每个阶段都由Spark任务(一个执行单元)组成然后在每个Spark executor 上执行.每个任务映射到一个核在一个数据分区上工作(图2-5)。因此一个16核的执行器可以在16个或更多的分区上并行运行16个或更多的任务这使得Spark的任务执行并行度很高图2 - 5. 创建一个或多个任务以分发给 executor 的 Spark stage 转换、立即执行操作和延迟求值
分布式数据上的Spark操作可以分为两种类型: 转换和执行操作。顾名思义转换在不改变原始数据的情况下将Spark DataFrame转换为新的DataFrame从而使其具有不可变性。换句话说像select()或filter()这样的操作不会改变原始DataFrame;相反它将返回转换后的操作结果作为一个新的DataFrame。所有的转换都是延迟执行的它们的结果不是立即计算出来的而是作为一个转换关系被记录。这些记录允许Spark在稍后的执行计划中重新安排某些转换合并它们或者将转换优化到更有效的执行阶段。延迟计算是Spark延迟执行的策略直到一个执行操作被调用或数据被“使用”(从磁盘读取或写入磁盘)。执行操作触发所有转换记录的延迟计算。在图2-6中所有的转换T都被记录下来直到动作A被调用。每个转换T产生一个新的DataFrame。图2 - 6. 延迟转换和立即执行求值的操作延迟求值通过转换血缘关系和数据不变性提供了容错性允许Spark通过链式调用转换来优化查询。由于Spark在转换血缘关系中记录了每个转换并且dataframe在转换之间是不可变的因此它可以通过简单地重新执行血缘关系的记录来重现其原始状态从而在发生故障时提供弹性。下边列出了一些转换和操作的示例这些动作和转换构成了一个Spark查询计划在调用操作之前查询计划中不会执行任何内容。下面的示例用Python和Scala显示有两个转换——read()和filter()——和一个立即执行操作 count()。该操作触发了作为查询执行计划的一部分记录的所有转换的执行。在这个例子中在shell中执行filter .count()之前什么都不会发生:
# In Python strings spark.read.text(../README.md)filtered strings.filter(strings.value.contains(Spark))filtered.count()
20// In Scala
scala import org.apache.spark.sql.functions._
scala val strings spark.read.text(../README.md)
scala val filtered strings.filter(col(value).contains(Spark))
scala filtered.count()
res5: Long 20s窄变换和宽变换
如前所述转换是Spark 延迟计算的操作。延迟求值方案的一个巨大优势是Spark可以检查你的计算性查询并确定如何优化它。这种优化可以通过连接或管道化一些操作并将它们分配到一个阶段来完成或者通过确定哪些操作需要跨集群的shuffle或数据交换来将它们分解为阶段来完成。转换可以分为窄依赖关系和宽依赖关系。任何可以从单个输入分区计算单个输出分区的转换都是窄转换。例如在前面的代码片段中filter()和contains()表示狭窄的转换因为它们可以在单个分区上操作并生成结果输出分区而无需交换任何数据。但是groupBy()或orderBy()之类的转换会指示Spark执行宽转换其中来自其他分区的数据被读入、合并并写入磁盘。如果我们要通过调用.orderby()对前面示例中filtered后的DataFrame进行排序那么每个分区都将在本地排序但是我们需要强制对集群中每个执行器分区中的数据进行过滤以便对所有记录进行排序。与窄转换相比宽转换需要其他分区的输出来计算最终的聚合。下图说明了两种类型的依赖关系
Spark UI
Spark包含一个图形用户界面可以使用它来检查或监视Spark应用程序的各个分解阶段(即job、state 和 tasks)。根据Spark的部署方式驱动程序启动一个web UI默认在端口4040上运行可以在其中查看指标和详细信息例如:
调度 stages 和 tasks 的列表RDD大小和内存使用的概要描述运行环境相关信息正在运行的 executors 信息所有的Spark SQL 查询
在本地模式下可以通过浏览器http://:4040访问该接口。 NOTE 当启动spark-shell时输出日志部分会显示要在端口4040上访问的本地主机URL。 让我们看一下前边的Python示例是如何转换为job、stage 和 tasks的。要查看DAG的外观单击web UI中的“DAG可视化”。如下图所示Driver 创建了一个 job 和一个 stage注意这里不需要Exchange(执行器之间交换数据的地方)因为只有一个阶段。每个单独的操作用蓝框表示。stage 0由一个task 组成。如果你有多个任务它们将并行执行。在“stages”页中可以查看各个stage的详细信息如下图所示在后续的文章系列中我会详细介绍 UI 界面的使用这里先只做个简单的介绍。
单机的应用程序
Spark发行版为每个Spark组件提供了一组示例应用程序。从本地机器上的安装目录中可以运行使用该命令提供的几个Java或Scala示例程序之一bin/run-example _class [params]_
$ ./bin/run-example JavaWordCount README.md这将在控制台上INFO 信息中输出 README.md 文件中每个单词的列表及其计数(计数单词是分布式计算的“Hello, World”)。
计算巧克力豆的数量
在前面的例子中我们统计了文件中的单词。如果文件很大它将分布在一个被划分为小数据块的集群上我们的Spark程序将分配计算每个分区中每个单词的任务并返回最终的聚合计数但这个例子已经有点过时了。让我们来解决一个类似的问题但是使用更大的数据集并使用更多Spark的分发功能和DataFrame api。如下图有很多巧克力豆的饼干我们需要将这些不同颜色的巧克力豆分配给不同的人。让我们编写一个Spark程序读取一个包含超过100,000个条目的文件(其中每行或每行都有一个state, mnm_color, count)并计算和汇总每种颜色和状态的计数。这些汇总的计数告诉我们每个人喜欢的mm巧克力豆的颜色。下边给出了完整的 Python 代码
# Import the necessary libraries.
# Since we are using Python, import the SparkSession and related functions
# from the PySpark module.
import sysfrom pyspark.sql import SparkSessionif __name__ __main__:if len(sys.argv) ! 2:print(Usage: mnmcount file, filesys.stderr)sys.exit(-1)# Build a SparkSession using the SparkSession APIs.# If one does not exist, then create an instance. There# can only be one SparkSession per JVM.spark (SparkSession.builder.appName(PythonMnMCount).getOrCreate())# Get the MM data set filename from the command-line argumentsmnm_file sys.argv[1]# Read the file into a Spark DataFrame using the CSV# format by inferring the schema and specifying that the# file contains a header, which provides column names for comma-# separated fields.mnm_df (spark.read.format(csv) .option(header, true) .option(inferSchema, true) .load(mnm_file))# We use the DataFrame high-level APIs. Note# that we dont use RDDs at all. Because some of Sparks # functions return the same object, we can chain function calls.# 1. Select from the DataFrame the fields State, Color, and Count# 2. Since we want to group each state and its MM color count,# we use groupBy()# 3. Aggregate counts of all colors and groupBy() State and Color# 4 orderBy() in descending ordercount_mnm_df (mnm_df.select(State, Color, Count).groupBy(State, Color).sum(Count).orderBy(sum(Count), ascendingFalse))# Show the resulting aggregations for all the states and colors;# a total count of each color per state.# Note show() is an action, which will trigger the above# query to be executed.count_mnm_df.show(n60, truncateFalse)print(Total Rows %d % (count_mnm_df.count()))# While the above code aggregated and counted for all # the states, what if we just want to see the data for # a single state, e.g., CA? # 1. Select from all rows in the DataFrame# 2. Filter only CA state# 3. groupBy() State and Color as we did above# 4. Aggregate the counts for each color# 5. orderBy() in descending order # Find the aggregate count for California by filteringca_count_mnm_df (mnm_df.select(State, Color, Count).where(mnm_df.State CA).groupBy(State, Color).sum(Count).orderBy(sum(Count), ascendingFalse))# Show the resulting aggregation for California.# As above, show() is an action that will trigger the execution of the# entire computation. ca_count_mnm_df.show(n10, truncateFalse)# Stop the SparkSessionspark.stop()创建 mnmcount.py 文件mnn_datasets .csv 文件数据集下载地址并使用安装的bin目录中的submit- Spark脚本将其作为Spark作业提交。将SPARK_HOME环境变量设置为在本地机器上安装Spark的根目录。 NOTE 前面的代码使用DataFrame API读起来像高级DSL查询。我将在后续文章中介绍这个和其他api。与RDD API不同你可以使用它来指示Spark做什么而不是如何做这是清晰和简单的! 为了避免将详细的INFO消息打印到控制台中请复制 log4j.properties.template 模板文件到 log4j.properties。并设置log4j.conf/log4j.conf文件中的rootCategoryWARN。执行提交命令提交上边的 Pyhton 代码至 Spark 集群
$SPARK_HOME/bin/spark-submit mnmcount.py data/mnm_dataset.csv---------------------
|State| Color|sum(Count)|
---------------------
| CA|Yellow| 100956|
| WA| Green| 96486|
| CA| Brown| 95762|
| TX| Green| 95753|
| TX| Red| 95404|
| CO|Yellow| 95038|
| NM| Red| 94699|
| OR|Orange| 94514|
| WY| Green| 94339|
| NV|Orange| 93929|
| TX|Yellow| 93819|
| CO| Green| 93724|
| CO| Brown| 93692|
| CA| Green| 93505|
| NM| Brown| 93447|
| CO| Blue| 93412|
| WA| Red| 93332|
| WA| Brown| 93082|
| WA|Yellow| 92920|
| NM|Yellow| 92747|
| NV| Brown| 92478|
| TX|Orange| 92315|
| AZ| Brown| 92287|
| AZ| Green| 91882|
| WY| Red| 91768|
| AZ|Orange| 91684|
| CA| Red| 91527|
| WA|Orange| 91521|
| NV|Yellow| 91390|
| UT|Orange| 91341|
| NV| Green| 91331|
| NM|Orange| 91251|
| NM| Green| 91160|
| WY| Blue| 91002|
| UT| Red| 90995|
| CO|Orange| 90971|
| AZ|Yellow| 90946|
| TX| Brown| 90736|
| OR| Blue| 90526|
| CA|Orange| 90311|
| OR| Red| 90286|
| NM| Blue| 90150|
| AZ| Red| 90042|
| NV| Blue| 90003|
| UT| Blue| 89977|
| AZ| Blue| 89971|
| WA| Blue| 89886|
| OR| Green| 89578|
| CO| Red| 89465|
| NV| Red| 89346|
| UT|Yellow| 89264|
| OR| Brown| 89136|
| CA| Blue| 89123|
| UT| Brown| 88973|
| TX| Blue| 88466|
| UT| Green| 88392|
| OR|Yellow| 88129|
| WY|Orange| 87956|
| WY|Yellow| 87800|
| WY| Brown| 86110|
---------------------Total Rows 60---------------------
|State| Color|sum(Count)|
---------------------
| CA|Yellow| 100956|
| CA| Brown| 95762|
| CA| Green| 93505|
| CA| Red| 91527|
| CA|Orange| 90311|
| CA| Blue| 89123|
---------------------首先我们看到每个地区的人喜欢的颜色的聚合数据下边是单个地区的。下边是 Scala 版本代码运行相同的应用程序
package main.scala.chapter2import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._/*** Usage: MnMcount mnm_file_dataset*/
object MnMcount {def main(args: Array[String]) {val spark SparkSession.builder.appName(MnMCount).getOrCreate()if (args.length 1) {print(Usage: MnMcount mnm_file_dataset)sys.exit(1)}// Get the MM data set filenameval mnmFile args(0)// Read the file into a Spark DataFrameval mnmDF spark.read.format(csv).option(header, true).option(inferSchema, true).load(mnmFile)// Aggregate counts of all colors and groupBy() State and Color// orderBy() in descending orderval countMnMDF mnmDF.select(State, Color, Count).groupBy(State, Color).sum(Count).orderBy(desc(sum(Count)))// Show the resulting aggregations for all the states and colorscountMnMDF.show(60)println(sTotal Rows ${countMnMDF.count()})println()// Find the aggregate counts for California by filteringval caCountMnNDF mnmDF.select(State, Color, Count).where(col(State) CA).groupBy(State, Color).sum(Count).orderBy(desc(sum(Count)))// Show the resulting aggregations for CaliforniacaCountMnMDF.show(10)// Stop the SparkSessionspark.stop()}
}单机编译 Scala 程序
下边将说明如何使用Scala构建工具(sbt)构建一个Scala Spark程序。build.sbt 是规范文件与makefile类似它描述并指示Scala编译器构建与Scala相关的任务例如 jar 包、packages、要解析的依赖项以及在哪里查找它们。下边是一个简单构建的例子
// Name of the package
name : main/scala/chapter2
// Version of our package
version : 1.0
// Version of Scala
scalaVersion : 2.12.10
// Spark library dependencies
libraryDependencies Seq(org.apache.spark %% spark-core % 3.0.0-preview2,org.apache.spark %% spark-sql % 3.0.0-preview2
)确保已经安装了Java开发工具包(JDK)和sbt并设置了JAVA_HOME和SPARK_HOME用一个命令就可以构建Spark应用程序:
$ sbt clean package
[info] Updated file /Users/julesdamji/gits/LearningSparkV2/chapter2/scala/
project/build.properties: set sbt.version to 1.2.8
[info] Loading project definition from /Users/julesdamji/gits/LearningSparkV2/
chapter2/scala/project
[info] Updating
[info] Done updating.
...
[info] Compiling 1 Scala source to /Users/julesdamji/gits/LearningSparkV2/
chapter2/scala/target/scala-2.12/classes ...
[info] Done compiling.
[info] Packaging /Users/julesdamji/gits/LearningSparkV2/chapter2/scala/target/
scala-2.12/main-scala-chapter2_2.12-1.0.jar ...
[info] Done packaging.
[success] Total time: 6 s, completed Jan 11, 2020, 4:11:02 PM成功构建后您可以运行Scala版本的计数示例如下所示:
$SPARK_HOME/bin/spark-submit --class main.scala.chapter2.MnMcount \
jars/main-scala-chapter2_2.12-1.0.jar data/mnm_dataset.csv
...
...
20/01/11 16:00:48 INFO TaskSchedulerImpl: Killing all running tasks in stage 4:
Stage finished
20/01/11 16:00:48 INFO DAGScheduler: Job 4 finished: show at MnMcount.scala:49,
took 0.264579 s
----------------
|State| Color|Total|
----------------
| CA|Yellow| 1807|
| CA| Green| 1723|
| CA| Brown| 1718|
| CA|Orange| 1657|
| CA| Red| 1656|
| CA| Blue| 1603|
----------------总结
在本章中我们介绍了开始使用Apache Spark所需的三个简单步骤:下载框架熟悉Scala或PySpark交互shell掌握高级Spark应用程序概念和术语。我们快速概述了使用转换和操作来编写Spark应用程序的过程并简要介绍了使用Spark UI来检查所创建的job、stage和task。最后通过一个简短的示例展示了如何使用高级结构化api来告诉Spark要做什么——在下一篇文章我将更详细地介绍这些api。
