spark学习笔记

1、目标

  • 1、熟悉spark相关概念
  • 2、搭建spark集群
  • 3、编写简单的spark程序
2、spark概述
  • spark是基于内存的一个计算框架,计算速度非常的快。这里面没有涉及到任何存储,如果想要处理外部的数据源,比如数据在HDFS上,此时我们就需要先搭建一个hadoop集群。

3、spark的特点

  • 1、速度快(比mapreduce在内存中快100倍,比在磁盘中快10倍)

    • (1)spark在处理的数据中间结果数据可以不落地,mapreduce每次中间结果都要落地。
    • (2)在mapreduce计算的时候,mapTask,reduceTask,每一个task都对应一个jvm进程。
      在spark中,它同样会按照hadoop中切片逻辑,会有N个task,而这些task都是运行在worker节点 上,worker上会有executor进程,而这些task会以线程的方式运行在executor上面。
  • 2、易用性

    • 可以使用多种语言来编写spark应用程序
      • java
      • scala
      • Python
      • R
  • 3、通用性

    • 可以使用sparksql、sparkStreaming、Mlib、Graphx
  • 4、兼容性

    • 可以运行在不同的资源调度平台
      • yarn(resourceManger分配资源)
      • mesos(是apache下开源的资源调度框架)
      • standAlone(master进行资源的分配)

4、spark集群安装

  • 1、下载对应版本的安装包
  • 2、上传安装包到服务器上
  • 3、规划一下安装目录
  • 4、解压安装包到指定的安装目录
  • 5、重命名安装目录
  • 6、修改配置文件 cd conf
    • (1) spark-env.sh.template (需要 mv spark-env.sh.template spark-env.sh)
      • 配置javahome export JAVA_HOME=/export/servers/jdk
      • 配置master的Host export SPARK_MASTER_HOST=node1
      • 配置master的Port export SPARK_MASTER_PORT=7077
    • (2)slaves.template (需要 mv slaves.template slaves)
      • 添加worker节点
        • node2
        • node3
  • 7、配置一下spark的环境变量
  • 8、通过scp命令分发到其他节点中
    • spark安装目录
    • /etc/profile
  • 9、所有机器都要source /etc/profile
  • 10、可以启动spark集群
    • $SPARK_HOME/sbin/start-all.sh
    • 可以通过web界面访问master
      • http://node1:8080
  • 11、停止spark集群
    • $SPARK_HOME/sbin/stop-all.sh

5、spark高可用集群配置

  • 1、需要先zk集群
    *2、修改spark配置(spark-env.sh)
    • (1)注释掉master的地址
    • (2) 引入zk配置
      • export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=hdp-node-01:2181,hdp-node-02:2181,hdp-node-03:2181 -Dspark.deploy.zookeeper.dir=/spark"
  • 3、启动
    • 启动zk集群
    • 需要在spark集群中任意一台机器上启动 start-all.sh
      • 产生master进程
      • 并且会根据 slaves,去对应的主机名上启动worker进程
    • 在其他worker节点上单独启动master
      • start-master.sh

6、初识spark程序

  • 已经知道那个master是活着的master
    • --master spark://node1:7077
  • 有很多的master时候
    • --master spark://node1:7077,node2:7077,node3:7077

7、spark-shell使用

  • 1、spark-shell --master local[N] (本地单机版)
    • local[N]:表示在本地模拟N个线程来运行当前任务
  • 2、spark-shell --master local[*] (本地单机版)
    • 这个*表示当前机器上所有可用的资源
  • 3、spark-shell --master spark://node2:7077
  • 4、spark-shell 读取hdfs上的数据文件
    • sc.textFile("hdfs://node1:9000/wc.txt").flatMap(.split(" ")).map(x=>(x,1)).reduceByKey(+_).collect

8、spark整合hdfs

  • 1、修改配置文件(spark-env.sh)
    • 添加配置参数
      • export HADOOP_CONF_DIR=/export/servers/hadoop/etc/hadoop
        *通过scp分发配置到其他节点
  • 2、可以sc.textFile("/wc.txt").flatMap(.split(" ")).map(x=>(x,1)).reduceByKey(+_).collect

9、scala语言编程spark单词计数

package cn.itcast.wc

import org.apache.spark.{SparkConf, SparkContext}

import org.apache.spark.rdd.RDD

//todo:通过scala编写spark的单词计数程序

object WordCount {

def main(args: Array[String]): Unit = {

//todo:1、创建SparkConf对象,设置appName和master地址,local[2]表示本地使用2个线程来进行计算

val sparkConf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local[2]")

//todo:2、创建SparkContext对象,这个对象很重要,它会创建DAGScheduler和TaskScheduler

val sc = new SparkContext(sparkConf)

//设置日志输出级别

sc.setLogLevel("WARN")

//todo:3、读取数据文件

val data: RDD[String] = sc.textFile("d:\data\in\words.txt")

//todo:4、切分每一行,并且压平  hello、you、me

val words: RDD[String] = data.flatMap(_.split(" "))

//todo:5、每个单词记位1     (hello,1)(hello,1)(hello,1)(you,1)(me,1)(me,1)

val wordAndOne: RDD[(String, Int)] = words.map((_,1))

//todo:6、相同单词出现的次数进行累加   (hello,10)(you,20)....

val result: RDD[(String, Int)] = wordAndOne.reduceByKey(+)

//todo:按照单词出现的次数降序排序

val sortResult: RDD[(String, Int)] = result.sortBy(.2,false)

//todo:7、收集数据,打印输出

val finalresult: Array[(String, Int)] = sortResult.collect()

//todo:打印结果

finalresult.foreach(x=>println(x))

//关闭

sc.stop()

}

}

10、java语言编程spark单词计数

package cn.itcast.wc;
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.Iterator;
import java.util.List;
    //todo:利用java语言来实现spark的单词计数
    public class WordCount_Java {
    public static void main(String[] args) {
        //todo:1、创建SparkConf对象,设置appName和master地址
        SparkConf sparkConf = new SparkConf().setAppName("WordCount_Java").setMaster("local[2]");
        
        //todo:2、创建javaSparkContext对象
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        
        //todo:3、读取数据文件
        JavaRDD<String> dataJavaRDD = jsc.textFile("D:\\data\\in\\words.txt");
        
        //todo:4、对每一行进行切分压平
        JavaRDD<String> wordsJavaRDD = dataJavaRDD.flatMap(new FlatMapFunction<String, String>() {
            @Override               //line表示每一行记录
            public Iterator<String> call(String line) throws Exception {
                //切分每一行
                String[] words = line.split(" ");
        
                return Arrays.asList(words).iterator();
            }
        });
        
        //todo:5、每个单词记为1
        JavaPairRDD<String, Integer> wordAndOneJavaPairRDD = wordsJavaRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<>(word, 1);
            }
        });
        
        //todo:6、把相同单词出现的次数累加  (_+_)
        JavaPairRDD<String, Integer> resultJavaPairRDD = wordAndOneJavaPairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        
        //按照单词出现的次数降序排序
        //需要将(单词,次数)进行位置颠倒 (次数,单词)
        JavaPairRDD<Integer, String> sortJavaPairRDD = resultJavaPairRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(Tuple2<String, Integer> t) throws Exception {
                return new Tuple2<>(t._2, t._1);
            }
        }).sortByKey(false);
        
        //将(次数,单词)变为(单词,次数)
        JavaPairRDD<String, Integer> finalSortJavaPairRDD = sortJavaPairRDD.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<Integer, String> t) throws Exception {
                return new Tuple2<>(t._2, t._1);
            }
        });
        
        //todo:7、收集打印
        List<Tuple2<String, Integer>> finalResult = finalSortJavaPairRDD.collect();
        
        for(Tuple2<String, Integer> t:finalResult){
            System.out.println(t);
        }
        
        jsc.stop();
    }
    }
添加新评论