Spark shell简介

• 启动 Spark shell 进入 Spark 安装目录后执行 spark-shell - -master master就可以提交Spark任务
• Spark shell 的原理是把每一·行Scala代码编译成类，最终交由Spark执行

Master 地址的设置

Master的地址可以有如下几种设置方式

地址	解释
local[N]	使用 N 条 Worker 线程在本地运行
spark://host:port	在 Spark standalone中 运行，指定 Spark 集群的Master地址，端口默认为 7077
mesos://host:port	在 Apache Mesos 中运行，指定 Meso的地址
yarn	在 Yarn 中运行，Yarn 的地址由环境变量 HADOOP_CONF_DIR 来指定

 

编写Spark代码的两种方式

编写Spark代码的两种方式

• spark-shell

  Spark shell是Spark提供的一个基于Scala语言的交互式解释器，类似于Scala提供的交互式解释器，Spark shel也可以直接在Shell中编写代码执行 这种方式也比较重要，因为一般的数据分析任务可能需要探索着进行，不是一藏而就的，使用Spark shell先进行探索，当代码稳定以后，使用独立应用的方式来提交任务，这样是一个比较常见的流程

• spark-submit

  Spark submit是一个命令，用于提交Scala编写的基于Spark框架，这种提交方式常用作于在集群中运行任务

 

Spark-Shell

Spark-Shell 读取本地文件

Step1 准备文件

在 master 中创建文件/root/data/wordcount.txt

vi /root/data/wordcount.txt
# 加入以下内容
hadoop spark flume
spark hadoop
flume hadoop

Step 2 启动Spark shell

cd /root/spark
bin/spark-shell --master local[2]

Step3 执行如下代码

scala> val sourceRdd = sc.textFile("file:///root/data/wordcount.txt")         # 读取文件
sourceRdd: org.apache.spark.rdd.RDD[String] = file:///root/data/wordcount.txt MapPartitionsRDD[1] at textFile at <console>:24scala> val flattenCountRdd = sourceRdd.flatMap(_.split(" ")).map((_,1))       # 拆分单词，并给与每个单词的词频为1
flattenCountRdd: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[3] at map at <console>:26scala> val aggCountRdd = flattenCountRdd.reduceByKey(_ + _)                   # 词频聚合    
aggCountRdd: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[4] at reduceByKey at <console>:28scala> val result = aggCountRdd.collect
result: Array[(String, Int)] = Array((spark,2), (hadoop,3), (flume,2))

运行流程

1. flatMap(_ .split(" ")) **将数据转为数组的形式，并展平为多个数据
2. map( _ , 1) 将数据转换为元组的形式
3. reduceByKey(_ + _) 将数据以 key 值相同聚合

 

Spark-Shell 读取HDFS

Step1上传文件到 HDFS 中

cd /root/data
hdfs dfs -mkdir /dataset
hdfs dfs -put wordcount.txt /dataset/

Step2 在Spark shell中访问 HDFS

val sourceRdd = sc.textFile("hdfs://master:9000/dataset/wordcount.txt") # sc.textFile("hdfs:///dataset/wordcount.txt") 或者 sc.textFile("/dataset/wordcount.txt") val flattenCountRdd = sourceRdd.flatMap(_.split(" ")).map((_,1))val aggCountRdd = flattenCountRdd.reduceByKey(_ + _)val result = aggCountRdd.collect

独立应用编写

Step 1 创建工程

1. 创建 IDEA 工程
   1. Create New Project → Maven → Next
   2. Groupld: cn.itcast → Artifactld: spark
2. 增加 Scala 支持

   1. 右键点击工程目录

      

   2. 点击增加框架支持

      

   3. 选择Scala添加框架支持

      

Step 2 编写 Maven 配置文件 pom.xml

1. 找到工程目录下的 pom.xml文件，（无则新增）

   

2. 添加以下内容

   <?xml version="1.0" encoding="UTF-8"?>
   <project xmlns="<http://maven.apache.org/POM/4.0.0>"xmlns:xsi="<http://www.w3.org/2001/XMLSchema-instance>"xsi:schemaLocation="<http://maven.apache.org/POM/4.0.0> <http://maven.apache.org/xsd/maven-4.0.0.xsd>"><modelVersion>4.0.0</modelVersion><groupId>cn.itcast</groupId><artifactId>spark</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties><dependencies><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>2.11.8</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.11</artifactId><version>2.1.1</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>2.7.7</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.10</version><scope>provided</scope></dependency></dependencies><build><sourceDirectory>src/main/scala</sourceDirectory><testSourceDirectory>src/test/scala</testSourceDirectory><plugins><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.2.2</version><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals><configuration><args><arg>-dependencyfile</arg><arg>${project.build.directory}/.scala_dependencies</arg></args></configuration></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>2.4.3</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><filters><filter><artifact>*:*</artifact><excludes><exclude>META-INF/*.SF</exclude><exclude>META-INF/*.DSA</exclude><exclude>META-INF/*.RSA</exclude></excludes></filter></filters><transformers><transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"></transformer></transformers></configuration></execution></executions></plugin></plugins></build>
   </project>
   

3. 创建目录 src/main/scala 和目录 src/test/scala

4. 创建Scala object WordCount

Step 3 编写代码

• 本地运行

  package cn.itcast.spark.rddimport org.apache.spark.{SparkConf, SparkContext}object WordCount {def main(args: Array[String]): Unit = {// 1、创建SparkContextval conf = new SparkConf().setMaster("local[6]").setAppName("word_count")val sc = new SparkContext(conf)// 2、加载文件//    1、准备文件//    2、读取文件val rdd1 = sc.textFile("./dataset/wordcount.txt")// 3、处理//    1、把整句话拆分成多个单词val rdd2 = rdd1.flatMap(item => item.split(" "))//    2、把每个单词指定一个词频1val rdd3 = rdd2.map(item => (item,1))//    3、整合val rdd4 = rdd3.reduceByKey((x , y) => (x + y))// 4、得到结果val result = rdd4.collect()result.foreach(item=>println(item))}
  }
  

• 提交运行

  package cn.itcast.spark.rddimport org.apache.spark.{SparkConf, SparkContext}object WordCount {def main(args: Array[String]): Unit = {// 1、创建SparkContextval conf = new SparkConf().setAppName("word_count")val sc = new SparkContext(conf)// 2、加载文件//    1、准备文件//    2、读取文件val rdd1 = sc.textFile("hdfs:///data/wordcount.txt")// 3、处理//    1、把整句话拆分成多个单词val rdd2 = rdd1.flatMap(item => item.split(" "))//    2、把每个单词指定一个词频1val rdd3 = rdd2.map(item => (item,1))//    3、整合val rdd4 = rdd3.reduceByKey((x , y) => (x + y))// 4、得到结果val result = rdd4.collect()result.foreach(item=>println(item))}
  }
  

Step 提交jar，虚拟机运行

cd /root/spark
bin/spark-submit --class cn.itcast.spark.rdd.WordCount --master spark://master:7077 /root/spark.jar