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>:24
scala> val flattenCountRdd = sourceRdd.flatMap(_.split(" ")).map((_,1)) # 拆分单词，并给与每个单词的词频为1
flattenCountRdd: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[3] at map at <console>:26
scala> val aggCountRdd = flattenCountRdd.reduceByKey(_ + _) # 词频聚合
aggCountRdd: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[4] at reduceByKey at <console>:28
scala> val result = aggCountRdd.collect
result: Array[(String, Int)] = Array((spark,2), (hadoop,3), (flume,2))
运行流程
- flatMap(_ .split(" ")) **将数据转为数组的形式，并展平为多个数据
- map( _ , 1) 将数据转换为元组的形式
- 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 创建工程
-
创建 IDEA 工程
- Create New Project → Maven → Next
- Groupld: cn.itcast → Artifactld: spark
-
增加 Scala 支持
-
右键点击工程目录
-
点击增加框架支持
-
选择Scala添加框架支持
-
Step 2 编写 Maven 配置文件 pom.xml
-
找到工程目录下的 pom.xml文件，（无则新增）
-
添加以下内容
<?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> -
创建目录 src/main/scala 和目录 src/test/scala
-
创建Scala object WordCount
Step 3 编写代码
-
本地运行
package cn.itcast.spark.rdd import org.apache.spark.{SparkConf, SparkContext} object WordCount { def main(args: Array[String]): Unit = { // 1、创建SparkContext val 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、把每个单词指定一个词频1 val 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)) } } -
提交运行文章来源：https://www.toymoban.com/news/detail-785949.html
package cn.itcast.spark.rdd import org.apache.spark.{SparkConf, SparkContext} object WordCount { def main(args: Array[String]): Unit = { // 1、创建SparkContext val 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、把每个单词指定一个词频1 val 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，虚拟机运行文章来源地址https://www.toymoban.com/news/detail-785949.html
cd /root/spark
bin/spark-submit --class cn.itcast.spark.rdd.WordCount --master spark://master:7077 /root/spark.jar
到了这里，关于Spark入门案例的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
