[spark]在Docker下安裝Spark（偽集群，只限於local模式）

找了很多文章，沒有詳細說明如何在docker上搭建spark，寫一篇隨筆做記錄

 

一，搭建spark

二，運行一個wordcount

 

 

硬件：centos 8.0 64位 阿里雲ECS服務器

 

安裝docker-compose

sudo pip install docker-compose==1.4.0

 

拉取鏡像：

docker pull singularities/spark:latest

我這里使用的是：singularities/spark這個鏡像  https://registry.hub.docker.com/r/singularities/spark

 

創建目錄

mkdir /root/docker-spark

 

創建對應master的卷目錄

mkdir /root/docker-spark/data-volumes-master

 

創建對應worker的卷目錄

mkdir /root/docker-spark/data-volumes-worker

 

創建docker-compose.yml

version: "2"

services:
  master:
    image: singularities/spark
    command: start-spark master
    hostname: master
    ports:
      - "6066:6066"
      - "7070:7070"
      - "8080:8080"
      - "50070:50070"
    volumes: 
      - /root/docker-spark/data-volumes-master:/data-volumes
  worker:
    image: singularities/spark
    command: start-spark worker master
    environment:
      SPARK_WORKER_CORES: 1
      SPARK_WORKER_MEMORY: 2g
    links:
      - master
    volumes: 
      - /root/docker-spark/data-volumes-worker:/data-volumes

 

有如下：　

 

 

 

建立JavaWordCount.java

/**
 * @ClassName JavaWordCount
 * @Author JK
 * @Description
 * @Date 2020-11-19 13:17
 * @Version 1.0
 */
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
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 org.apache.spark.sql.SparkSession;
import scala.Tuple2;

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

/**
 * Created by xiaosi on 17-2-13.
 *
 * Spark 測試程序 WordCount
 *
 */

public final class JavaWordCount {

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

        SparkSession spark = SparkSession.builder().appName("JavaWordCount").getOrCreate();

        JavaRDD<String> lines = spark.read().textFile(args[0]).javaRDD();

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

        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());
        }
        spark.stop();
    }
}

 

其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>org.jk</groupId>
    <artifactId>TestSpark</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <spark.version>2.2.1</spark.version>
    </properties>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>6</source>
                    <target>6</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

    <dependencies>

        <dependency> <!-- Spark dependency -->
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
    </dependencies>
</project>

 

build成jar包：略

 

然后建立word.txt做數據（自己喜歡怎么填都行，空格隔開）

aa bb cc dd we adrq gdsgwet sda aweq

 

打包好的jar包和txt數據，上傳到之前的master卷中

 

 

 

使用docker exec 命令進入master容器

 

 

 

我們在hadoop中的根目錄下創建目錄data，並把work.txt上傳到hadoop里面

#建立目錄
hadoop fs -mkdir /data
#上傳文件
hadoop fs -put word.txt hdfs://master:8020/data/word.txt

　　

在 yourserver's ip:50070下的browse directory可以看到word.txt

 

 

 

然后在你的bash下執行

spark-submit --master spark://master:7077 --name MyWordCount --class JavaWordCount /data-volumes/TestSpark.jar hdfs://master:8020/data/word.txt --executor-memory 512M --executor-core 1 --num-executors 1 --driver-cores 1

 

spark-summit的參數參考：

--master     master 的地址，提交任務到哪里執行，例如 spark://host:port,  yarn,  local
--deploy-mode    在本地 (client) 啟動 driver 或在 cluster 上啟動，默認是 client
--class  應用程序的主類，僅針對 java 或 scala 應用
--name   應用程序的名稱
--jars   用逗號分隔的本地 jar 包，設置后，這些 jar 將包含在 driver 和 executor 的 classpath 下
--packages   包含在driver 和executor 的 classpath 中的 jar 的 maven 坐標
--exclude-packages   為了避免沖突 而指定不包含的 package
--repositories   遠程 repository
--conf PROP=VALUE   指定 spark 配置屬性的值，例如 -conf spark.executor.extraJavaOptions="-XX:MaxPermSize=256m"
--properties-file    加載的配置文件，默認為 conf/spark-defaults.conf
--driver-memory  Driver內存，默認 1G
--driver-java-options    傳給 driver 的額外的 Java 選項
--driver-library-path    傳給 driver 的額外的庫路徑
--driver-class-path  傳給 driver 的額外的類路徑
--driver-cores   Driver 的核數，默認是1。在 yarn 或者 standalone 下使用
--executor-memory    每個 executor 的內存，默認是1G
--total-executor-cores   所有 executor 總共的核數。僅僅在 mesos 或者 standalone 下使用
--num-executors  啟動的 executor 數量。默認為2。在 yarn 下使用
--executor-core  每個 executor 的核數。在yarn或者standalone下使用

　　

你就會在你的 ip:8080看到正在running的application

 

 

 

但是！！！！！！還是被執行時kill掉

 

 

 

如果把命令的參數從 --master spark://master:7077 換成 --master local

spark-submit --master local --name MyWordCount --class JavaWordCount /data-volumes/TestSpark.jar hdfs://master:8020/data/word.txt --executor-memory 512M --executor-core 1 --num-executors 1 --driver-cores 1

 

即可運行順利

 

 

分析原因：

local模式下  驅動程序driver就是執行了一個Spark Application的main函數和創建Spark Context的進程，它包含了這個application的全部代碼。

而如果在standalone下，估計是Spark Application 檢測到1個core，導致在Registered StateStoreCoordinator endpoint就發生錯誤，kill掉了

具體原因仍在排查，歡迎指出不對！

補充原因：

以下是錯誤日志：

 

內存狀態：初步估計內存問題，待有機會升級服務器再嘗試解決問題