Apache Spark 2.2.0 中文文档 - 快速入门 | ApacheCN

快速入门

本教程提供了如何使用 Spark 的快速入门介绍。首先通过运行 Spark 交互式的 shell(在 Python 或 Scala 中)来介绍 API, 然后展示如何使用 Java , Scala 和 Python 来编写应用程序。

为了继续阅读本指南, 首先从 Spark 官网 下载 Spark 的发行包。因为我们将不使用 HDFS, 所以你可以下载一个任何 Hadoop 版本的软件包。

请注意, 在 Spark 2.0 之前, Spark 的主要编程接口是弹性分布式数据集(RDD)。 在 Spark 2.0 之后, RDD 被 Dataset 替换, 它是像RDD 一样的 strongly-typed(强类型), 但是在引擎盖下更加优化。 RDD 接口仍然受支持, 您可以在 RDD 编程指南 中获得更完整的参考。 但是, 我们强烈建议您切换到使用 Dataset(数据集), 其性能要更优于 RDD。 请参阅 SQL 编程指南 获取更多有关 Dataset 的信息。

使用 Spark Shell 进行交互式分析

基础

Spark shell 提供了一种来学习该 API 比较简单的方式, 以及一个强大的来分析数据交互的工具。在 Scala(运行于 Java 虚拟机之上, 并能很好的调用已存在的 Java 类库)或者 Python 中它是可用的。通过在 Spark 目录中运行以下的命令来启动它:

./bin/spark-shell

Spark 的主要抽象是一个称为 Dataset 的分布式的 item 集合。Datasets 可以从 Hadoop 的 InputFormats(例如 HDFS文件)或者通过其它的 Datasets 转换来创建。让我们从 Spark 源目录中的 README 文件来创建一个新的 Dataset:

scala> val textFile = spark.read.textFile("README.md")
textFile: org.apache.spark.sql.Dataset[String] = [value: string]

您可以直接从 Dataset 中获取 values(值), 通过调用一些 actions(动作), 或者 transform(转换)Dataset 以获得一个新的。更多细节, 请参阅 API doc

scala> textFile.count() // Number of items in this Dataset
res0: Long = 126 // May be different from yours as README.md will change over time, similar to other outputs

scala> textFile.first() // First item in this Dataset
res1: String = # Apache Spark

现在让我们 transform 这个 Dataset 以获得一个新的。我们调用 filter 以返回一个新的 Dataset, 它是文件中的 items 的一个子集。

scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))
linesWithSpark: org.apache.spark.sql.Dataset[String] = [value: string]

我们可以链式操作 transformation(转换)和 action(动作):

scala> textFile.filter(line => line.contains("Spark")).count() // How many lines contain "Spark"?
res3: Long = 15

Dataset 上的更多操作

Dataset actions(操作)和 transformations(转换)可以用于更复杂的计算。例如, 统计出现次数最多的单词 :

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)
res4: Long = 15

第一个 map 操作创建一个新的 Dataset, 将一行数据 map 为一个整型值。在 Dataset 上调用 reduce 来找到最大的行计数。参数 map 与 reduce 是 Scala 函数(closures), 并且可以使用 Scala/Java 库的任何语言特性。例如, 我们可以很容易地调用函数声明, 我们将定义一个 max 函数来使代码更易于理解 :

scala> import java.lang.Math
import java.lang.Math

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => Math.max(a, b))
res5: Int = 15

一种常见的数据流模式是被 Hadoop 所推广的 MapReduce。Spark 可以很容易实现 MapReduce:

scala> val wordCounts = textFile.flatMap(line => line.split(" ")).groupByKey(identity).count()
wordCounts: org.apache.spark.sql.Dataset[(String, Long)] = [value: string, count(1): bigint]

在这里, 我们调用了 flatMap 以 transform 一个 lines 的 Dataset 为一个 words 的 Dataset, 然后结合 groupByKey 和 count 来计算文件中每个单词的 counts 作为一个 (String, Long) 的 Dataset pairs。要在 shell 中收集 word counts, 我们可以调用 collect:

scala> wordCounts.collect()
res6: Array[(String, Int)] = Array((means,1), (under,2), (this,3), (Because,1), (Python,2), (agree,1), (cluster.,1), ...)

缓存

Spark 还支持 Pulling(拉取)数据集到一个群集范围的内存缓存中。例如当查询一个小的 “hot” 数据集或运行一个像 PageRANK 这样的迭代算法时, 在数据被重复访问时是非常高效的。举一个简单的例子, 让我们标记我们的 linesWithSpark 数据集到缓存中:

scala> linesWithSpark.cache()
res7: linesWithSpark.type = [value: string]

scala> linesWithSpark.count()
res8: Long = 15

scala> linesWithSpark.count()
res9: Long = 15

使用 Spark 来探索和缓存一个 100 行的文本文件看起来比较愚蠢。有趣的是, 即使在他们跨越几十或者几百个节点时, 这些相同的函数也可以用于非常大的数据集。您也可以像 编程指南. 中描述的一样通过连接 bin/spark-shell 到集群中, 使用交互式的方式来做这件事情。

独立的应用

假设我们希望使用 Spark API 来创建一个独立的应用程序。我们在 Scala(SBT), Java(Maven)和 Python 中练习一个简单应用程序。

我们将在 Scala 中创建一个非常简单的 Spark 应用程序 - 很简单的, 事实上, 它名为 SimpleApp.scala:

/* SimpleApp.scala */
import org.apache.spark.sql.SparkSession

object SimpleApp {
  def main(args: Array[String]) {
    val logFile = "YOUR_SPARK_HOME/README.md" // Should be some file on your system
    val spark = SparkSession.builder.appName("Simple Application").getOrCreate()
    val logData = spark.read.textFile(logFile).cache()
    val numAs = logData.filter(line => line.contains("a")).count()
    val numBs = logData.filter(line => line.contains("b")).count()
    println(s"Lines with a: $numAs, Lines with b: $numBs")
    spark.stop()
  }
}

注意, 这个应用程序我们应该定义一个 main() 方法而不是去扩展 scala.App。使用 scala.App 的子类可能不会正常运行。

该程序仅仅统计了 Spark README 文件中每一行包含 ‘a’ 的数量和包含 ‘b’ 的数量。注意, 您需要将 YOUR_SPARK_HOME 替换为您 Spark 安装的位置。不像先前使用 spark shell 操作的示例, 它们初始化了它们自己的 SparkContext, 我们初始化了一个 SparkContext 作为应用程序的一部分。

我们调用 SparkSession.builder 以构造一个 [[SparkSession]], 然后设置 application name(应用名称), 最终调用 getOrCreate 以获得 [[SparkSession]] 实例。

我们的应用依赖了 Spark API, 所以我们将包含一个名为 build.sbt 的 sbt 配置文件, 它描述了 Spark 的依赖。该文件也会添加一个 Spark 依赖的 repository:

name := "Simple Project"

version := "1.0"

scalaVersion := "2.11.8"

libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.2.0"

为了让 sbt 正常的运行, 我们需要根据经典的目录结构来布局 SimpleApp.scala 和 build.sbt 文件。在成功后, 我们可以创建一个包含应用程序代码的 JAR 包, 然后使用 spark-submit 脚本来运行我们的程序。

# Your directory layout should look like this
$ find .
.
./build.sbt
./src
./src/main
./src/main/scala
./src/main/scala/SimpleApp.scala

# Package a jar containing your application
$ sbt package
...
[info] Packaging {..}/{..}/target/scala-2.11/simple-project_2.11-1.0.jar

# Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit \
  --class "SimpleApp" \
  --master local[4] \
  target/scala-2.11/simple-project_2.11-1.0.jar
...
Lines with a: 46, Lines with b: 23

快速跳转

恭喜您成功的运行了您的第一个 Spark 应用程序!

# 针对 Scala 和 Java, 使用 run-example:
./bin/run-example SparkPi

# 针对 Python 示例, 直接使用 spark-submit:
./bin/spark-submit examples/src/main/python/pi.py

# 针对 R 示例, 直接使用 spark-submit:

./bin/spark-submit examples/src/main/r/dataframe.R

我们一直在努力

原文地址: http://spark.apachecn.org/docs/cn/2.2.0/quick-start.html 网页地址: http://spark.apachecn.org/ github: https://github.com/apachecn/spark-doc-zh(觉得不错麻烦给个 Star,谢谢!~)

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏向治洪

数据结构之哈希函数

概念: 哈希(hash),也叫做散列、数据摘要等,是一种常见的数据结构。哈希的表的核心概念分为哈希表和哈希函数。 哈希表(hashTable) 哈希表之前讲过,...

2077
来自专栏向治洪

Volley请求

1. Volley简介 我们平时在开发Android应用的时候不可避免地都需要用到网络技术,而多数情况下应用程序都会使用HTTP协议来发送和接收网络数据。A...

1857
来自专栏PHP在线

PHP使用SOAP调用.net的WebService问题

项目的需求,需要和一个.net系统进行数据交换,合作方提供了一个WebService接口。这个与一般的PHP POST或GET传值再查库拿数据的思路有点不一样,...

3156
来自专栏C#

ASP.NET的session操作方法总结

    在开发ASP.NET程序时,需要对相关数据进行缓存,缓存较多的主要是用户的身份信息,现提供几个对session操作较为常用的方法: 1.添加sess...

2197
来自专栏用户2442861的专栏

python操作redis简单例子

#2、配置  配置一下吧,默认配置文件在: “/etc/redis/redis.conf”  绑定ip:  “bind 127.0.0.1″ -> “b...

592
来自专栏Golang语言社区

【转】Go语言Http Server源码阅读

目录(?)[-] 前言 几个重要概念 具体分析 几个接口 Handler ResponseWriter Flusher Hijacker response H...

2724
来自专栏刘望舒

设计模式(十五)状态模式

前言 建议在阅读本文前先阅读设计模式(十一)策略模式这篇文章,虽说状态模式和策略模式的结构几乎是相同的,但是它们所解决的问题是不同的,读完这两篇文章你就会有了答...

1846
来自专栏皮振伟的专栏

[THP][redis]THP对redis的影响

前言: 前文《[linux][redis]bgsave引起的latency突刺问题分析》分析了redis-server执行bgsave因为fork引起的late...

1282
来自专栏Kubernetes

原 荐 Kubernetes Resourc

更多关于kubernetes的深入文章,请看我csdn或者oschina的博客主页。 ResoureQuota介绍 关于ResoureQuota和Resourc...

4659
来自专栏SDNLAB

SDN实战团分享(七):YANG模型与OpenDaylight南北向接口

YANG模型是什么? YANG模型是一种数据建模语言,用来建模由NETCONF协议、NETCONF远端过程调用(RPCs)、和NETCONF通知(notific...

3938

扫码关注云+社区