码迷,mamicode.com
首页 > 其他好文 > 详细

Spark编程基础

时间:2019-06-20 11:07:51      阅读:105      评论:0      收藏:0      [点我收藏+]

标签:使用   filter   pre   amp   line   数据源   ado   htm   RKE   

ISBN 9787115488169

之前我们已经完成了Hadoop+Spark集群的搭建,下面改用这本书。

教材讲义实验

sbt

Spark应用程序开发可以采用Scala+sbt,Java+Maven,或Python直接spark-submit三种方式。

这里介绍通过sbt编译打包Scala程序的命令行方法,也可以使用Intellij idea或Eclipse等集成开发环境。

安装

echo "deb https://dl.bintray.com/sbt/debian /" | sudo tee -a /etc/apt/sources.list.d/sbt.list
sudo apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 2EE0EA64E40A89B84B2DF73499E82A75642AC823
sudo apt-get update
sudo apt-get install sbt

Scala独立应用程序

项目结构:

mkdir -p ./sparkapp/src/main/scala

sparkapp/src/main/scala目录下新建SimpleApp.scala,该程序计算/usr/local/spark/README.md文件中包含a的行数和包含b的行数。

/* 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 = "file:///usr/local/spark/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))
    }
}

sparkapp目录下新建simple.sbt

name := "Simple Project"
version := "1.0"
scalaVersion := "2.11.8"
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.3"

使用sbt编译打包:

sbt package

首次使用要从网络上下载依赖包,可能会比较久。

将生成的jar包通过spark-submit提交到Spark中运行:

spark-submit --class "SimpleApp" target/scala-2.11/simple-project_2.11-1.0.jar 2>&1 | grep "Lines with a:"

HDFS文件存储架构

在开始分布式程序设计之前,用一个实例来认识一下分布式环境中数据的分布。

配置三个slave服务器(即3个datanode),dfs.replication为2,block size为128M的条件下,假设我们有一个490M的文件,则在HDFS中会被切分成4个block(0, 1, 2, 3),每个block复制2份,在50070可以查看block id,从而在slave的dfs.datanode.data.dir目录下可以看到具体的区块分布情况。比如在我的测试中,slave1上有上述文件的block 1, 2, 3,slave2上有block 0,slave3上有block 0, 1, 2, 3。

因此数据的不完整导致了集中式算法不能得到正确的结果,需要设计并行算法。当然实际上并不关心数据的具体分布情况,这里只是给出一个直观的感受。

启动Spark集群

在master和slave服务器上修改spark-env.sh

export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)
export HADOOP_CONF_DIR=/usr/local/hadoop/etc/hadoop
export SPARK_MASTER_IP=172.17.0.5

启动Hadoop集群:start-all.sh

启动Spark集群:

cd /usr/local/spark/sbin
./start-master.sh
./start-slaves.sh

使用jps可以看到master上多了一个Master进程,slave上多了一个Worker进程。也可以通过http://master:8080查看。

Spark RDD

Resilient Distributed Dataset(弹性分布式数据集)是Spark的核心,属于分布式的内存系统的数据集应用,可以与外部存储系统的数据集兼容,如HDFS、HBase或其他Hadoop数据源。

RDD之上有三种基本运算:

  1. 转换:会产生新的RDD,但这类运算不会立刻执行,而是等到执行动作运算时一起执行
  2. 动作:不产生新的RDD,而是产生数值、数组或写入文件系统
  3. 持久化:将重复使用的RDD持久化在内存中

RDD特性:

  1. immutable:不可变
  2. lineage:记录每个RDD与其父代RDD之间的关联,通过什么操作得到该RDD

因此Spark具备容错特性,某节点机器故障,RDD损毁,会重新执行一连串转换命令,产生新的输出数据。

Scala集合操作

  1. 遍历
  2. map,flatmap
  3. filter
  4. reduce
  5. fold

Spark编程基础

标签:使用   filter   pre   amp   line   数据源   ado   htm   RKE   

原文地址:https://www.cnblogs.com/humz/p/10935903.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!