Spark编程基础

标签：使用   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