IDEA连接spark集群

写在前面

我所使用的spark集群是基于docker搭建的，一共三个节点，一个master和两个slave，安装过程参考我之前的博客https://blog.csdn.net/weixin_43622131/article/details/110098712
宿主机使用的是windows10专业版

一、安装Scala插件

搜索Scala，下载如图所示的插件，因为是从国外源下载，速度比较慢，所以建议科学上网下载

二、新建一个Maven项目

自己填写GroupId和ArtifactId

三、编写pom.xml文件

里面的spark插件版本和hadoop插件版本要和spark集群的版本相对应，避免出现问题

<?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.example</groupId>
    <artifactId>spark</artifactId>
    <version>1.0-SNAPSHOT</version>


    <properties>
        <scala.version>2.11.8</scala.version>
        <hadoop.version>2.7.4</hadoop.version>
    </properties>

    <repositories>
        <repository>
            <id>scala-tools.org</id>
            <name>Scala-Tools Maven2 Repository</name>
            <url>http://scala-tools.org/repo-releases</url>
        </repository>
    </repositories>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.10</artifactId>
            <version>2.4.7</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.10</artifactId>
            <version>2.4.7</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.10</artifactId>
            <version>2.4.7</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
    </dependencies>

</project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

四、导入Scala环境

在windows中如果没有Scala环境则先安装Scala环境
由于安装包下载很慢我这里直接给出安装包，版本是2.11.8（要与spark集群的Scala版本一致）
链接：https://pan.baidu.com/s/1I62fSLbr2NakycxioTlopw
提取码：oq2j
选择Project Structure

在Libraries里导入Scala SDK环境

选择Scala的安装路径即可识别Scala环境

在src/main下创建scala目录
在Project Structure下的Modules中将scala文件夹设置成source文件夹

五、导入spark的jar包

我是将spark的包直接解压，然后将其中的jars文件夹导入，如果不做这一步运行程序的时候会出现jar包找不到的错误，导入spark的jar包就可以了

六、编写Scala程序并提交给集群

在scala目录下创建一个Scala类，我所创建的是SparkPi2类作为例子，是用来计算圆周率的

/**
 * Created by zf on 12/3/20.
 */
import scala.math.random

import org.apache.spark._

/** Computes an approximation to pi */
object SparkPi2 {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("Spark Pi").set("spark.executor.memory", "512m")
      .set("spark.driver.host","10.0.75.1")//这个ip很重要，我因为这个ip没有设置正确卡了好长时间，我使用的是docker，这个ip就要设置为本机在docker分配的虚拟网卡中的ip地址，如果设置成其他网卡的ip会被主机拒绝访问
      .set("spark.driver.cores","1")
      .setMaster("spark://127.0.0.1:7077") //这里应设为master的ip加上配置spark时设置的端口，一般都为7077，因为我使用的是docker所以将master的端口映射到本地了，所以直接用127.0.0.1来访问
      .setJars(List("D:\\自然语言处理\\spark\\out\\artifacts\\SparkExample_jar\\spark.jar"))

    val spark = new SparkContext(conf)
    val slices = if (args.length > 0) args(0).toInt else 2
    val n = 100000 * slices
    val count = spark.parallelize(1 to n, slices).map { i =>
      val x = random * 2 - 1
      val y = random * 2 - 1
      if (x * x + y * y < 1) 1 else 0
    }.reduce(_ + _)
    println("Pi is roughly " + 4.0 * count / n)
    spark.stop()
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

再强调下这个参数，如果是docker搭建的spark
一定要使用docker分配的虚拟网卡的ip！！！
一定要使用docker分配的虚拟网卡的ip！！！
一定要使用docker分配的虚拟网卡的ip！！！

set("spark.driver.host","10.0.75.1")
1

如果没有设置这个参数或者设置的ip错误就会出现下列错误，这个错误困扰了我很久，终于在改成了docker分配虚拟网卡的ip后得到解决
看一下运行的结果吧：
成功连接到了spark集群

两个节点在执行各自的task

运行出结果

---

如果要读取文件的话，是读取不了本地的或者集群上本地文件的，会报找不到文件，所以必须将文件上传到hdfs上通过hdfs进行读取。
这里读取hdfs，如果使用的是docker搭建的集群则无法直接访问，需要将9000端口映射到本地，然后使用docker网络给本地分配的IP进行访问，因为使用spark集群运行程序要保证每个节点都能访问数据，所以要使windows能被所有节点访问，则需要关闭windows的防火墙，就可以成功访问hdfs文件了。

import java.io.FileWriter
import java.net.InetAddress

import org.apache.spark._


object test4  {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("test2").set("spark.executor.memory", "512m")
      .set("spark.driver.host", "10.0.75.1")
      .set("spark.driver.cores", "2")
      .setMaster("spark://127.0.0.1:7077") //spark://127.0.0.1:7077
      .setJars(List("D:\\自然语言处理\\spark\\out\\artifacts\\SparkExample_jar\\spark.jar")) // maven打的jar包的路径
      .set("spark.driver.allowMultipleContexts", "true")
    val sc = new SparkContext(conf)
    // 读取文件A
    var A = sc.textFile("hdfs://10.0.75.1:9000/test2/Data01.txt") // 关闭windows的防火墙
    A.collect().foreach(e => println(e))
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

成功运行