如何使用Java調用Spark集群

更新時間：2024年02月29日 09:45:26 作者：bluesnail95

這篇文章主要介紹了如何使用Java調用Spark集群,我搭建的Spark集群的版本是2.4.4,本文結合示例代碼給大家介紹的非常詳細,感興趣的朋友跟隨小編一起看看吧

我搭建的Spark集群的版本是2.4.4。

在網上找的maven依賴，鏈接忘記保存了。。。。

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <hadoop.version>2.6.0-cdh5.14.2</hadoop.version>
    <hive.version>1.1.0-cdh5.14.2</hive.version>
    <hbase.version>1.2.0-cdh5.14.2</hbase.version>
    <scala.version>2.11.8</scala.version>
    <spark.version>2.4.4</spark.version>
</properties>
<repositories>
    <repository>
        <id>cloudera</id>
        <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
    </repository>
</repositories>
<dependencies>
    <!--scala-->
    <dependency>
        <groupId>org.scala-lang</groupId>
        <artifactId>scala-library</artifactId>
        <version>2.11.8</version>
    </dependency>
    <!-- spark-core -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>${spark.version}</version>
    </dependency>
    <!-- spark-sql -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.11</artifactId>
        <version>${spark.version}</version>
    </dependency>
    <!-- spark-hive -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-hive_2.11</artifactId>
        <version>2.4.4</version>
    </dependency>
    <!-- spark-graphx -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-graphx_2.11</artifactId>
        <version>${spark.version}</version>
    </dependency>
    <!-- hadoop -->
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>${hadoop.version}</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-hdfs</artifactId>
        <version>${hadoop.version}</version>
    </dependency>
    <!-- log4j -->
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
    </dependency>
    <!-- junit -->
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.11</version>
    </dependency>
    <!-- kafka-clients -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.11.0.2</version>
    </dependency>
    <!-- mysql-connector-java -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.31</version>
    </dependency>
</dependencies>
<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <version>2.0.1.RELEASE</version>
            <configuration>
                <mainClass>gdut.spark.SparkInit</mainClass>
            </configuration>
            <executions>
                <execution>
                    <goals>
                        <goal>repackage</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

Java客戶端連接示例：

import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.SparkConf;
import java.util.Arrays;
import java.util.List;
public class SparkInit {
    public static void main(String[] args) {
        try {
            SparkConf conf = new SparkConf().setAppName("liufeifei").setMaster("spark://x.x.x.x:30010");
            conf.set("spark.executor.cores","1");
            conf.set("spark.executor.memory", "1024m");
            JavaSparkContext sc = new JavaSparkContext(conf);
            List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
            JavaRDD<Integer> distData = sc.parallelize(data);
            System.out.println("result is " + distData.count());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

遇到問題：
(1)spark集群中，worker節(jié)點提示：Failed to send RPC
master pod的spark-shell執(zhí)行collect方法，日志輸出如下：

worker pod輸出如下：

worker節(jié)點無法創(chuàng)建Executor，在worker節(jié)點的安裝目錄下有個work目錄，有每次創(chuàng)建Executor的日志。查看是worker節(jié)點與master節(jié)點無法通信。但是worker節(jié)點有向master注冊，在master的UI界面有顯示注冊的worker節(jié)點。在網上不經意看到有人說可能是istio影響了，然后想起自己之前部署過istio。查看spark部署的命名空間確實是開啟istio注入。

換個沒有istio注入的命名空間創(chuàng)建spark集群。在master節(jié)點的spark-shell可以執(zhí)行collect方法，可以調度到worker節(jié)點的Executor。

(2)Caused by: java.net.UnknownHostException: XXX
無論在本地還是在虛擬機執(zhí)行上面的客戶端連接，都會提示UnknownHostException。這是因為在worker容器的/etc/hosts找不到客戶端主機名稱和IP的映射關系。

解決辦法:使用 HostAliases 向 Pod /etc/hosts 文件添加條目

hostAliases:
  - ip: "127.0.0.1"
    hostnames:
    - "foo.local"
    - "bar.local"
  - ip: "10.1.2.3"
    hostnames:
    - "foo.remote"
    - "bar.remote"

我在yaml文件添加了hostAliases之后，提示主機名不符合規(guī)定，然后修改了自己虛擬機上的主機名。

LINUX修改主機名稱(立即永久生效)

修改主機名后遇到：java.net.UnknownHostException:Name or Service not known

修改了/etc/hosts文件可以解決。

因為spark集群是部署在一臺虛擬機上，本地不能和虛擬機通信，所以要把spring boot項目打包成jar在虛擬機上執(zhí)行。
Main方法輸出：