这期内容当中小编将会给大家带来有关如何进行Spark 1.6.0 新手的快速入门,文章内容丰富且以专业的角度为大家分析和叙述,阅读完这篇文章希望大家可以有所收获。
Spark交互式Shell的使用
基础
Spark的交互式Shell提供了一个简单的方式来学习Spark的API,同时也提供了强大的交互式数据处理能力。Spark Shell支持Scala和Python两种语言。启动支持Scala的Spark Shell方式为
./bin/spark-shell
Spark最重要的一个抽象概念是弹性分布式数据集(Resilient Distributed Dataset)简称RDD。RDDs可以通过Hadoop InputFormats(例如HDFS文件)创建,也可以由其它RDDs转换而来。下面的例子是通过加载Spark目录下的README.md文件生成 RDD的例子:
scala> val textFile = sc.textFile("README.md") textFile: spark.RDD[String] = spark.MappedRDD@2ee9b6e3
RDDs有两种操作:
-
actions:返回计算值
-
transformations:返回一个新RDDs的引用
actions示例如下:
scala> textFile.count() // Number of items in this RDD res0: Long = 126 scala> textFile.first() // First item in this RDD res1: String = # Apache Spark
如下transformations示例,使用filter操作返回了一个新的RDD,该RDD为文件中数据项的子集,该子集符合过滤条件:
scala> val linesWithSpark = textFile.filter(line => line.contains("Spark")) linesWithSpark: spark.RDD[String] = spark.FilteredRDD@7dd4af09
Spark也支持将actions和transformations一起使用:
scala> textFile.filter(line => line.contains("Spark")).count() // How many lines contain "Spark"? res3: Long = 15
更多RDD操作(More on RDD Operations)
RDD的actions和transformations操作可以用于更加复杂的计算。下面是查找README.md文件中单词数最多的行的单词数目:
scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b) res4: Long = 15
上面代码中,***个map操作将一行文本按空格分隔,并计算单词数目,将line映射为一个integer值,并创建了一个新的RDD保存这些 integer值。RDD调用reduce计算***的单词数。示例中map和reduce操作的参数是Scala的函数式编程风格,Spark支持 Scala、Java、Python的编程风格,并支持Scala/Java库。例如,使用Scala中的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
随着Hadoop的流行,MapReduce变为一种常见的数据流模式。Spark可以轻松的实现MapReduce,使用Spark编写MapReduce程序更加简单:
scala> val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b) wordCounts: spark.RDD[(String, Int)] = spark.ShuffledAggregatedRDD@71f027b8
上面示例中,使用flatMap、map和reduceByKey操作来计算每个单词在文件中出现的次数,并生成一个结构为的RDD。可以使用collect操作完成单词统计结果的收集整合:
scala> wordCounts.collect() res6: Array[(String, Int)] = Array((means,1), (under,2), (this,3), (Because,1), (Python,2), (agree,1), (cluster.,1), ...)
缓存
Spark支持将数据缓存到集群的分布式内存中。在数据会被重复访问的情况下,将数据缓存到内存能减少数据访问时间,从而提高运行效率。尤其是在数据分布在几十或几百个节点上时,效果更加明显。下面为将数据linesWithSpark缓存到内存的示例:
scala> linesWithSpark.cache() res7: spark.RDD[String] = spark.FilteredRDD@17e51082 scala> linesWithSpark.count() res8: Long = 19 scala> linesWithSpark.count() res9: Long = 19
独立应用
假设我们想使用Spark API编写独立应用程序。我们可以使用Scala、Java和Python轻松的编写Spark应用。下面示例为一个简单的应用示例:
-
Scala
/* SimpleApp.scala */ import org.apache.spark.SparkContext import org.apache.spark.SparkContext._ import org.apache.spark.SparkConf object SimpleApp { def main(args: Array[String]) { val logFile = "YOUR_SPARK_HOME/README.md" // Should be some file on your system val conf = new SparkConf().setAppName("Simple Application") val sc = new SparkContext(conf) val logData = sc.textFile(logFile, 2).cache() val numAs = logData.filter(line => line.contains("a")).count() val numBs = logData.filter(line => line.contains("b")).count() println("Lines with a: %s, Lines with b: %s".format(numAs, numBs)) } }
上面程序分别统计了README中包含字符‘a’以及‘b’的行数。与前面Spark shell例子不同的是,我们需要初始化SparkContext。
我们通过SparkContext创建了一个SparkConf对象,SparkConf对象包含应用的基本信息。
我们基于Spark API编写应用,所以我们需要编写一个名为“simple.sbt”的sbt配置文件,用于指明Spark为该应用的一个依赖。下面的sbt配置文件示例中,还增加了Spark的一个依赖库“spark-core”:
name := "Simple Project" version := "1.0" scalaVersion := "2.10.5" libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"
为了让sbt正确执行,我们需要对SimpleApp.scala和simple.sbt根据sbt要求的目录结构布局。如果布局正确,就可以生成该应用的JAR包,使用spark-submit命令即可运行该程序。
-
Javaga
/* SimpleApp.java */ import org.apache.spark.api.java.*; import org.apache.spark.SparkConf; import org.apache.spark.api.java.function.Function; public class SimpleApp { public static void main(String[] args) { String logFile = "YOUR_SPARK_HOME/README.md"; // Should be some file on your system SparkConf conf = new SparkConf().setAppName("Simple Application"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDDlogData = sc.textFile(logFile).cache(); long numAs = logData.filter(new Function
该示例的代码逻辑同上一段Scala示例代码。与Scala示例类似,首先初始化了SparkContext,通过SparkContext创建了JavaSparkContext对象。并创建了RDDs以及执行transformations操作。***,通过继承了spark.api.java.function.Function的类将函数传给Spark。
在这里,使用Maven进行编译,Maven的pom.xml如下:
<project> <groupId>edu.berkeley</groupId> <artifactId>simple-project</artifactId> <modelVersion>4.0.0</modelVersion> <name>Simple Project</name> <packaging>jar</packaging> <version>1.0</version> <dependencies> <dependency> <!-- Spark dependency --> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.10</artifactId> <version>1.6.0</version> </dependency> </dependencies> </project>
按照Maven的要求架构配置文件位置:
$ find . ./pom.xml ./src ./src/main ./src/main/java ./src/main/java/SimpleApp.java
现在,就可以使用Maven打包应用,以及使用命令./bin/spark-submit.执行该应用程序。示例如下:
# Package a JAR containing your application $ mvn package ... [INFO] Building jar: {..}/{..}/target/simple-project-1.0.jar # Use spark-submit to run your application $ YOUR_SPARK_HOME/bin/spark-submit / --class "SimpleApp" / --master local[4] / target/simple-project-1.0.jar ... Lines with a: 46, Lines with b: 23
上述就是小编为大家分享的如何进行Spark 1.6.0 新手的快速入门了,如果刚好有类似的疑惑,不妨参照上述分析进行理解。如果想知道更多相关知识,欢迎关注亿速云行业资讯频道。
原创文章,作者:3628473679,如若转载,请注明出处:https://blog.ytso.com/220408.html