Spark Shell 终端操作指南

曾经终败给现在 2023-06-18 11:54 18阅读 0赞

## 基础使用 ##

vim /opt/spark-2.1.1-bin-hadoop2.7/conf/spark-env.sh
    
    向文件中添加这么一行，标识本地集群
    SPARK_LOCAL_IP=192.168.78.135

使用以下命令启动 spark shell 终端

[root@single bin]# cd /opt/spark-2.1.1-bin-hadoop2.7/bin
    [root@single bin]# ./spark-shell local
    Using Spark's 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).
    19/12/01 19:59:01 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
    19/12/01 19:59:05 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
    Spark context Web UI available at http://192.168.78.135:4040
    Spark context available as 'sc' (master = local[*], app id = local-1575259142063).
    Spark session available as 'spark'.
    Welcome to
          ____              __
         / __/__  ___ _____/ /__
        _\ \/ _ \/ _ `/ __/  '_/
       /___/ .__/\_,_/_/ /_/\_\   version 2.1.1
          /_/
             
    Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_191)
    Type in expressions to have them evaluated.
    Type :help for more information.
    
    scala>

Spark的主要抽象是一个称为弹性分布式数据集(RDD)的分布式项集合。通过读取本地文件或者其他 RDD 来创建新的 RDD。下面通过读取本地文件的方式创建 RDD。

scala> val textFile = sc.textFile("README.md")
    textFile: org.apache.spark.rdd.RDD[String] = README.md MapPartitionsRDD[1] at textFile at <console>:25

RDD 有 action 算子来打印输出数据，transformation 算子来通过一个 RDD 生成另一个新的 RDD。下面是几个简单的 action 算子。

scala> textFile.count() // 几条数据
    res0: Long = 126 // May be different from yours as README.md will change over time, similar to other outputs
    
    scala> textFile.first() // 输出打印第一条数据
    res1: String = # Apache Spark

下面是一个简单的 transformation 算子，用来筛选出包含 “Spark” 的数据

scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))
    linesWithSpark: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at filter at <console>:27

可以在 transformation 算子之后使用 action 算子，打印输出结果。

scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))
    linesWithSpark: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at filter at <console>:27

## 其他 RDD 算子 ##

action 和 transformation 可以用来更复杂的计算，假设我们想找到单词最多的一行：

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

map 算子将数据以空格切分，数据为一个 int 类型 的 List<Int>，Math.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

一种常见的数据流模式是 MapReduce，它是由 Hadoop 普及的。Spark 可以轻松实现 MapReduce 流：

scala> val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)
    wordCounts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[8] at reduceByKey at <console>:28

在这里，我们将 flatMap、map 和 reduceByKey 算子组合在一起，以(String、Int)对的 RDD 的形式计算文件中的每个单词的计数。要收集 shell 中的字数，可以使用 collect 操作：

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

## 缓存 ##

也可以将数据缓存到内存中，用来频繁查询或计算的数据。使用如下：

scala> linesWithSpark.cache()
    res7: linesWithSpark.type = MapPartitionsRDD[2] at filter at <console>:27
    
    scala> linesWithSpark.count()
    res8: Long = 15
    
    scala> linesWithSpark.count()
    res9: Long = 15

## 独立程序开发样例 ##

import org.apache.spark.SparkContext
    import org.apache.spark.SparkContext._
    import org.apache.spark.SparkConf
    
    object SimpleApp {
      def main(args: Array[String]) {
        val logFile = "/opt/spark-2.1.1-bin-hadoop2.7/README.md" // 对应自己 spark 的目录
        val conf = new SparkConf().setAppName("Simple Application")
        val sc = new SparkContext(conf)
        // 读取本地文件并缓存
        val logData = sc.textFile(logFile, 2).cache()
        // 将每一行带有 a 的进行聚合
        val numAs = logData.filter(line => line.contains("a")).count()
        // 同上，带 b 的
        val numBs = logData.filter(line => line.contains("b")).count()
        println(s"Lines with a: $numAs, Lines with b: $numBs")
        sc.stop()
      }
    }

> 注意：程序需要定义 main 方法。

与之前不同的是：我们需要自己创建 SparkContext。