1、spark概述

  • spark是基于内存的一个计算框架,计算速度非常的快。这里面没有涉及到任何存储,如果想要处理外部的数据源,比如数据在HDFS上,此时我们就需要先搭建一个hadoop集群。

2、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进行资源的分配)

3、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://node01:8080
  • 11、停止spark集群
    • $SPARK_HOME/sbin/stop-all.sh

4、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

5、初识spark程序

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

6、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

7、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

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

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
//通过scala编写spark的单词计数程序
object WordCount {

  def main(args: Array[String]): Unit = {
    //1、创建SparkConf对象,设置appName和master地址,local[2]表示本地使用2个线程来进行计算
    val sparkConf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local[2]")
    //2、创建SparkContext对象,这个对象很重要,它会创建DAGScheduler和TaskScheduler
    val sc: SparkContext = new SparkContext(sparkConf)
    //设置日志输出级别
    sc.setLogLevel("WARN")
    //3、读取数据文件
    //val data: RDD[String] = sc.textFile(args(0))
    val data: RDD[String] = sc.textFile("E:\\wordcount\\input\\words.txt")
    //4、切分每一行,并且压平  hello、you、me
    val words: RDD[String] = data.flatMap(_.split(" "))
    //5、每个单词记位1
    val wordAndOne: RDD[(String, Int)] = words.map((_,1))
    //6、相同单词出现的次数进行累加
    val result: RDD[(String, Int)] = wordAndOne.reduceByKey(_+_)
    //按照单词出现的次数降序排序
    val sortResult: RDD[(String, Int)] = result.sortBy(_._2,false)
    //7、收集数据,打印输出
    val finalResult: Array[(String, Int)] = sortResult.collect()
    //打印结果
    finalResult.foreach(x=>println(x))
    //关闭
    sc.stop()
  }
}

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

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;

//利用java语言来实现spark的单词计数
public class WordCount_Java {
    public static void main(String[] args) {
        //1、创建SparkConf对象,设置appName和master地址
        SparkConf sparkConf = new SparkConf().setAppName("WordCount_Java").setMaster("local[2]");

        //2、创建javaSparkContext对象
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);

        //3、读取数据文件
        JavaRDD<String> dataJavaRDD = jsc.textFile("E:\\wordcount\\input\\words.txt");

        //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();
            }
        });

        //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);
            }
        });

        //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);
            }
        });

        //7、收集打印
        List<Tuple2<String, Integer>> finalResult = finalSortJavaPairRDD.collect();

        for(Tuple2<String, Integer> t:finalResult){
            System.out.println(t);
        }

        jsc.stop();
    }
}

10、上面的程序所依赖的pom.xml文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>cn.itcast</groupId>
    <artifactId>Spark</artifactId>
    <version>1.0-SNAPSHOT</version>
    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>18</maven.compiler.target>
        <encoding>UTF-8</encoding>
        <scala.version>2.11.8</scala.version>
        <scala.compat.version>2.11</scala.compat.version>
        <hadoop.version>2.7.4</hadoop.version>
        <spark.version>2.0.2</spark.version>
    </properties>
    <dependencies>
    <dependency>
        <groupId>org.scala-lang</groupId>
        <artifactId>scala-library</artifactId>
        <version>${scala.version}</version>
    </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
    </dependencies>
    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>2.18.1</version>
                <configuration>
                    <useFile>false</useFile>
                    <disableXmlReport>true</disableXmlReport>
                    <includes>
                        <include>**/*Test.*</include>
                        <include>**/*Suite.*</include>
                    </includes>
                </configuration>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass></mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

Spark运行基本流程图

图片1.png

评论已关闭

暂无评论