spark - 如何写spark脚本以及简单运行

- 日理万妓 2022-08-31 15:24 403阅读 0赞

## spark - 快速开始 ##

首先，通过spark的交互式shell介绍API(Python/Scala)，然后介绍怎么使用java/python/scala来写应用程序。

首先从spark[官网下载][Link 1]spark的打包版本，由于我们不会使用 HDFS，您可以为任何版本的 Hadoop 下载一个包。

注意，在 Spark 2.0 之前，Spark 的主要编程接口是**弹性分布式数据集**（RDD）

> RDD：弹性分布式数据集

## RDD & DataSet ##

在 Spark 2.0 之后，RDD 被 Dataset 取代，它像 RDD 一样是强类型的，但在`hood`下有更丰富的优化。

仍然支持 RDD 接口，您可以在 [RDD 编程指南][RDD]中获得更详细的参考，但是，我们强烈建议您改用 **Dataset**，它的性能比 RDD 好。请参阅 [SQL 编程指南][SQL]以获取有关 Dataset 的更多信息。

## 安全性 ##

默认情况下，Spark 中的安全性是关闭的。 这可能意味着默认情况下您很容易受到攻击。 请在运行 Spark 之前查看 [Spark 安全性][Spark]。

# 使用 Spark Shell 进行交互式分析 #

## 基本操作 ##

Spark 的 shell 提供了一种简单的 API 学习方式，以及一种交互式分析数据的强大工具，它可以在 Scala（它在 Java VM 上运行，因此是使用现有 Java 库的好方法）或 Python 中可用，通过以下方式启动`spark shell`

### Scala启动方式 ###

暂略

### Python启动方式 ###

./bin/pyspark

或者，如果 PySpark 在您当前的环境中安装了 pip:

pyspark

Spark 的主要抽象是一个分布式项目集合，称为数据集。**数据集可以从 Hadoop InputFormats（例如 HDFS 文件）或通过转换其他数据集创建**。**由于 Python 的动态特性，我们不需要在 Python 中对数据集进行强类型化**。因此，Python 中所有的 Dataset 都是 **Dataset\[Row\]**，我们称其为 **DataFrame** 以与 Pandas 和 R 中的数据框概念保持一致。让我们从 Spark 源目录中的 README 文件的文本中创建一个新的 DataFrame：

>>> textFile = spark.read.text("README.md")

您可以通过调用某些操作直接从 DataFrame 获取值，或者转换 DataFrame 以获得新的值。 有关更多详细信息，请阅读 [API 文档][API]。

>>> textFile.count()
    108
    >>> textFile.first()
    Row(value='# Apache Spark')

现在让我们将这个 DataFrame 转换为一个新的。 我们调用 filter 来返回一个新的 DataFrame，其中包含文件中行的子集。

>>> linesWithSpark = textFile.filter(textFile.value.contains("Spark"))

我们可以将转换和操作链接在一起：

>>> textFile.filter(textFile.value.contains("Spark")).count()  # How many lines contain "Spark"?
    15

## 更多关于数据集的操作 ##

数据集操作和转换可用于更复杂的计算。 假设我们想找到**单词最多的那一行**：

>>> from pyspark.sql.functions import *
    >>> textFile.select(size(split(textFile.value, "\s+")).name("numWords")).agg(max(col("numWords"))).collect()
    [Row(max(numWords)=15)]

这首先将一行映射到一个整数值并将其别名为“numWords”，从而创建一个新的 `DataFrame`。 在该 `DataFrame` 上调用 `agg` 以查找最大字数。 `select` 和 `agg` 的参数都是 Column，我们可以使用 `df.colName` 从 `DataFrame` 中获取列。 我们还可以导入 `pyspark.sql.functions`，它提供了很多方便的函数来从旧的 Column 构建一个新的 Column。

一种常见的数据流模式是 `MapReduce`，由 `Hadoop` 推广。 Spark 可以轻松实现 MapReduce 流程：

>>> wordCounts = textFile.select(explode(split(textFile.value, "\s+")).alias("word")).groupBy("word").count()

在这里，我们使用 `select` 中的`explode`函数，将行的数据集转换为单词的数据集，然后结合 `groupBy` 和 `count` 来计算文件中的每个单词计数作为 2 列的 DataFrame：`word` & `count`

要在我们的 shell 中收集字数，我们可以调用 collect：

>>> wordCounts.collect()
    [Row(word=u'online', count=1), Row(word=u'graphs', count=1), ...]

## 缓存 ##

Spark 还支持将**数据集**拉入**集群范围的内存缓存**中，这在重复访问数据时非常有用，例如在查询一个小的“热”数据集或运行像 PageRank 这样的迭代算法时，作为一个简单的例子，让我们标记要缓存的 linesWithSpark 数据集：

>>> linesWithSpark.cache()
    
    >>> linesWithSpark.count()
    15
    
    >>> linesWithSpark.count()
    15

使用 Spark 浏览和缓存一个 100 行的文本文件似乎很愚蠢。有趣的是，这些相同的函数可以用于非常大的数据集，即使它们跨越数十或数百个节点，您还可以通过将 bin/pyspark 连接到集群以交互方式执行此操作，如 [RDD 编程指南][RDD 1]中所述。

## 自包含应用程序 ##

假设我们希望使用 Spark API 编写一个自包含的应用程序。 我们将在 **Scala（使用 sbt）**、**Java（使用 Maven）** 和 **Python（使用 pip）** 中演练一个简单的应用程序。

现在我们将展示如何使用 Python API (PySpark) 编写应用程序。  
如果您正在构建打包的 PySpark 应用程序或库，您可以将其添加到 setup.py 文件中，如下所示：

install_requires=[
            'pyspark==3.1.2'
        ]

例如，我们将创建一个简单的 Spark 应用程序 SimpleApp.py：

"""SimpleApp.py"""
    from pyspark.sql import SparkSession
    
    logFile = "YOUR_SPARK_HOME/README.md"  # Should be some file on your system
    spark = SparkSession.builder.appName("SimpleApp").getOrCreate()
    logData = spark.read.text(logFile).cache()
    
    numAs = logData.filter(logData.value.contains('a')).count()
    numBs = logData.filter(logData.value.contains('b')).count()
    
    print("Lines with a: %i, lines with b: %i" % (numAs, numBs))
    
    spark.stop()

该程序仅计算文本文件中包含“a”的行数和包含“b”的行数，与 **Scala** 和 **Java** 示例一样，我们使用 **SparkSession** 来创建数据集。对于使用自定义类或第三方库的应用程序，我们还可以通过其 **–py-files** 参数将代码依赖项添加到 `spark-submit`，方法是将它们打包到 .zip 文件中（有关详细信息，请参阅 `spark-submit --help`）。 SimpleApp 足够简单，我们不需要指定任何代码依赖项。

我们可以使用`bin/spark-submit` 脚本运行这个应用程序：

${SAPRK_HOME}/bin/spark-submit \
      --master local[4] \
      SimpleApp.py

如果在python环境中安装了`pyspark`(`pip install pyspark`)，可以使用python解释器运行程序，或者根据需要而使用`spark-submit`

# Use the Python interpreter to run your application
    $ python SimpleApp.py
    ...
    Lines with a: 46, Lines with b: 23

[Link 1]: https://spark.apache.org/downloads.html
[RDD]: https://spark.apache.org/docs/latest/rdd-programming-guide.html
[SQL]: https://spark.apache.org/docs/latest/sql-programming-guide.html
[Spark]: https://spark.apache.org/docs/latest/security.html
[API]: http://spark.apache.org/docs/latest/api/python/index.html#pyspark.sql.DataFrame
[RDD 1]: http://spark.apache.org/docs/latest/rdd-programming-guide.html#using-the-shell