spark学习笔记-二

spark学习笔记-2
本节记录spark下载-->编译-->安装-->使用

首先从主站下载spark源码包（源码包更利于我们学习spark）
http://spark.apache.org/downloads.html
注意选择所需要的相对应的spark源码版本，在此我们选择使用spark-1.3.0

spark的所有版本源码全部托管在 github上面
https://github.com/apache/spark
spark 源码包编译有三种方式
1：SBT编译
2：Maven编译
3：打包编译 make-distribution.sh

note:编译spark需要maven 3.0.4+   jdk1.6+，据说spark1.5之后jdk需要1.7+版本支持
编译之前，需要调整maven 缓存大小

export MAVEN_OPTS="-Xmx2g -XX:MaxPermSize=512M -XX:ReservedCodeCacheSize=512m"

maven编译参数设置:
1：hdfs版本参数设置

# Apache Hadoop 1.2.1
mvn -Dhadoop.version=1.2.1 -DskipTests clean package

# Cloudera CDH 4.2.0 with MapReduce v1
mvn -Dhadoop.version=2.0.0-mr1-cdh4.2.0 -DskipTests clean package

# Apache Hadoop 0.23.x
mvn -Phadoop-0.23 -Dhadoop.version=0.23.7 -DskipTests clean package

2：yarn框架支持

# Apache Hadoop 2.2.X
mvn -Pyarn -Phadoop-2.2 -Dhadoop.version=2.2.0 -DskipTests clean package

# Apache Hadoop 2.3.X
mvn -Pyarn -Phadoop-2.3 -Dhadoop.version=2.3.0 -DskipTests clean package

# Apache Hadoop 2.4.X or 2.5.X
mvn -Pyarn -Phadoop-2.4 -Dhadoop.version=VERSION -DskipTests clean package

Versions of Hadoop after 2.5.X may or may not work with the -Phadoop-2.4 profile (they were
released after this version of Spark).

# Different versions of HDFS and YARN.
mvn -Pyarn -Phadoop-2.3 -Dhadoop.version=2.3.0 -Dyarn.version=2.2.0 -DskipTests clean package

3：hive 和jdbc支持

# Apache Hadoop 2.4.X with Hive 13 support
mvn -Pyarn -Phadoop-2.4 -Dhadoop.version=2.4.0 -Phive -Phive-thriftserver -DskipTests clean package

# Apache Hadoop 2.4.X with Hive 12 support
mvn -Pyarn -Phadoop-2.4 -Dhadoop.version=2.4.0 -Phive -Phive-0.12.0 -Phive-thriftserver -DskipTests clean package

4：scala支持
注意：scala默认是2.10

spark源码包中提供了一个打包shell make-distribution.sh
我们只需要提供相应的参数，便可以完成打包

./make-distribution.sh --tgz -Pyarn -Phadoop-2.4 Dhadoop.version=2.4.0 -Phive-0.13.1 -Phive-thriftserver

在此需要注意一下,由于国内网络限制，所以请在maven setting.xml 中配置镜像地址

<mirror>
  <id>nexus-osc</id>
  <mirror>*</mirror>
  <name>Nexus osc</name>
  <url>http://maven.oschina.net/conyent/groups/public/</url>
</morrir>

配置google的域名解析器

 nameserver 8.8.8.8
 nameserver 8.8.4.4

执行编译脚本，会发现开始时间很长，原因是make-distribution.sh脚本中有一段代码是检测hive hadoop 版本信息，可以将这一段代码直接写死。

VERSION=$("$MVN" help:evaluate -Dexpression=project.version 2>/dev/null | grep -v "INFO" | tail -n 1)
SPARK_HADOOP_VERSION=$("$MVN" help:evaluate -Dexpression=hadoop.version $@ 2>/dev/null\
    | grep -v "INFO"\
    | tail -n 1)
SPARK_HIVE=$("$MVN" help:evaluate -Dexpression=project.activeProfiles -pl sql/hive $@ 2>/dev/null\
    | grep -v "INFO"\
    | fgrep --count "<id>hive</id>";\
    # Reset exit status to 0, otherwise the script stops here if the last grep finds nothing\
    # because we use "set -o pipefail"
    echo -n)

============改写成===========
VERSION=1.3.0 #spark版本
SPARK_HADOOP_VERSION=2.4.0 #hadoop版本
SPARK_HIVE=1 #支持hive

Standalone Mode配置方式
如下图为架构图




配置spark-env.sh文件 需要配置的参数如下

JAVA_HOME
SCALA_HOME
HADOOP_CONF_DIR
SPARK_MASTER_IP
SPARK_MASTER_PORT
SPARK_MASTER_WEBUI_PORT
SPARK_WORKER_CORES
SPARK_WORKER_PORT
SPARK_WORKER_WEBUI_PORT
SPARK_WORKER_INSTANCES

启动

./sbin/start-master.sh  //启动主节点
./sbin/start-slave.sh   //启动从节点
./bin/spark-shell --master spark://IP:PORT //启动application

测试运行实例

scala> val textFile = sc.textFile("README.md")  //加载文件到内存
       textFile.count() //计算行数
       textFile.first() //返回第一行
       //返回一个行包含Spark的字符数组
       scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))
//找出包含字节最多的行  
textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)
//单词统计
val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)
wordCounts.collect()

使用java 代码调用api

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);
    JavaRDD<String> logData = sc.textFile(logFile).cache();

    long numAs = logData.filter(new Function<String, Boolean>() {
      public Boolean call(String s) { return s.contains("a"); }
    }).count();

    long numBs = logData.filter(new Function<String, Boolean>() {
      public Boolean call(String s) { return s.contains("b"); }
    }).count();

    System.out.println("Lines with a: " + numAs + ", lines with b: " + numBs);
  }
}