标签:使用 filter pre amp line 数据源 ado htm RKE
ISBN 9787115488169
之前我们已经完成了Hadoop+Spark集群的搭建,下面改用这本书。
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
项目结构:
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:"
在开始分布式程序设计之前,用一个实例来认识一下分布式环境中数据的分布。
配置三个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。
因此数据的不完整导致了集中式算法不能得到正确的结果,需要设计并行算法。当然实际上并不关心数据的具体分布情况,这里只是给出一个直观的感受。
在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
查看。
Resilient Distributed Dataset(弹性分布式数据集)是Spark的核心,属于分布式的内存系统的数据集应用,可以与外部存储系统的数据集兼容,如HDFS、HBase或其他Hadoop数据源。
RDD之上有三种基本运算:
RDD特性:
因此Spark具备容错特性,某节点机器故障,RDD损毁,会重新执行一连串转换命令,产生新的输出数据。
标签:使用 filter pre amp line 数据源 ado htm RKE
原文地址:https://www.cnblogs.com/humz/p/10935903.html