Spark分布式环境搭建

Spark分布式环境搭建

1. 准备
Scala安装包：scala-2.11.7.tgz
SPARK安装包：spark-1.1.0-bin-hadoop2.4.tar.gz
2. 说明
默认已安装了hadoop环境，jdk，配置了免密码登录，设置了主机名。我们一共有三台机器，分别为：hadoop1，hadoop2，hadoop3。其中hadoop1为hadoop集群的master，其他两台是slave。Spark也装3台，其中hadoop1位master其余为worker。
3.
Scala安装
在主节点hadoop1中做一下操作（如果没有说明所有操作都是在主节点）
创建目录,并切换到该目录
cd /home/hadoop/scala
解压scala
tar –zxvf scala-2.11.7.tgz
将scala添加到环境变量
vi /etc/profile
添加一下内容：
export SCALA_HOME=/home/hadoop/scala/scala-2.11.7
修改PATH
export PATH=$PATH:$SCALA_HOME/bin
保存退出，使其生效
source /etc/profile
查看是否安装成功
scala –version
复制安装包到两个从节点（hadoop2，hadoop3）
scp /home/hadoop/scala/scala-2.11.7 root@hadoop2:/ /home/hadoop/scala
scp /home/hadoop/scala/scala-2.11.7 root@hadoop3:/ /home/hadoop/scala
按主节点的方式配置(hadoop2，hadoop3)上scala环境变量
4.
Spark安装
创建目录
mkdir /home/hadoop/spark
cd /home/hadoop/spark
解压spark文件
tar –zxvf spark-1.1.0-bin-hadoop2.4.tar.gz
设置spark环境变量
vi /etc/profile
添加如下内容
export SPARK_HOME=/home/hadoop/spark/spark-1.1.0-bin-hadoop2.4
修改PATH
export PATH=$PATH:$SCALA_HOME/bin:$SPARK_HOME/bin
使修改生效
source /etc/profile
修改spark配置文件spark-env.sh
cd $SPARK_HOME/conf
cp spark-env.sh.template spark-env.sh
vi spark-env.sh
添加以下内容
#jdk安装目录

export JAVA_HOME=/home/hadoop/java/jdk1.7.0_71

#scala安装目录

export SCALA_HOME=/home/hadoop/scala/scala-2.11.7
#spark集群的master节点的ip

export SPARK_MASTER_IP=192.168.0.2

#指定的worker节点能够最大分配给Excutors的内存大小

export SPARK_WORKER_MEMORY=1g

#hadoop集群的配置文件目录

export HADOOP_CONF_DIR=/usr/local/hadoop/hadoop-2.5.2/etc/hadoop

修改conf目录下面的slaves文件
vi slaves
添加一下内容
hadoop1

hadoop2

hadoop3

保存退出后对从节点（hadoop2，hadoop3）做同样的设置
5.
Spark启动
master主机${SPARK_HOME}/bin目录下运行start-all.sh启动集群
访问http://hadoop1:8080查看节点情况(默认是8080端口，如果不想使用8080端口，需要在spark-env.sh里面配置SPARK_MASTER_WEBUI_PORT来指定端口号)
Spark默认地址为：spark://masterIP:7077

6.
Spark-shell使用
cd $SPARK_HOME/bin
启动spark-shell
./spark-shell
测试数组相加：
scala> val data = Array(1, 2, 3, 4, 5) //产生data
scala> val distData = sc.parallelize(data) //将data处理成RDD
scala> distData.reduce(_+_) //在RDD上进行运算，对data里面元素进行加和
测试统计字符数：
scala> val distFile = sc.textFile("hdfs://hadoop1:9000/input/file01.txt")
scala> distFile.map(_.size).reduce(_+_)
查看job执行情况
http:masterIp:4040/stages
7. Java版wordCount
需要依赖jar包：$SPARK_HOME/lib/spark-assembly-1.1.0-hadoop2.4.0.jar

package spark.test;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;

public final class WordCount {
private static final Pattern SPACE = Pattern.compile(" ");

public static void main(String[] args) throws Exception {

if (args.length < 1) {
System.err.println("Usage: JavaWordCount <file>");
System.exit(1);
}

SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount");
JavaSparkContext ctx = new JavaSparkContext(sparkConf);
JavaRDD<String> lines = ctx.textFile(args[0], 1);

JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
@Override
public Iterable<String> call(String s) {
return Arrays.asList(SPACE.split(s));
}
});

JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {
@Override
public Tuple2<String, Integer> call(String s) {
return new Tuple2<String, Integer>(s, 1);
}
});

JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer i1, Integer i2) {
return i1 + i2;
}
});

List<Tuple2<String, Integer>> output = counts.collect();
for (Tuple2<?, ?> tuple : output) {
System.out.println(tuple._1() + ": " + tuple._2());
}
ctx.stop();
}
}
对以上代码打包：WordCount.jar到 home/test/
然后运行
spark-submit --master spark://hadoop1:7077 --name WordCount --class spark.test.WordCount --executor-memory 1G --total-executor-cores 2 /home/test/WordCount.jar hdfs://hadoop1:9000/home/input/mapengbo.txt