kubernetes Sping Boot Spark on K8S(Minikube)：无法分配java.lang.invoke.SerializedLambda的示例

我看到其他人也在处理同样的问题，但由于没有一个建议的解决方案或变通方法对我有效，而且我已经花了几个小时在这上面，我想我会详细分享我的具体案例，希望有人能指出我错过了什么。
我想尝试在Minikube k8s集群上运行一个非常简单的Spark Spring-Boot应用程序。当我在本地运行应用程序（使用SparkSession.builder().master("local")）时，一切都按预期工作。然而，当我将应用程序部署到minikube时，我设法让我的驱动程序pod在作业触发时启动执行器pod，但随后我在执行器pod上得到了这个异常：

ERROR Executor: Exception in task 0.1 in stage 0.0 (TID 1)
cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.sql.execution.MapPartitionsExec.func of type scala.Function1 in instance of org.apache.spark.sql.execution.MapPartitionsExec

这是我的spring-boot应用程序。为了简单起见，我将所有逻辑都保留在控制器上：

WordcountController

@RestController
public class WordCountController implements Serializable {
    @PostMapping("/wordcount")
    public ResponseEntity<String> handleFileUpload(@RequestParam("file") MultipartFile file) throws IOException {
        String hostIp;
        try {
            hostIp = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            throw new RuntimeException(e);
        }

        SparkConf conf = new SparkConf();
        conf.setAppName("count.words.in.file")
                .setMaster("k8s://https://kubernetes.default.svc:443")
                .setJars(new String[]{"/app/wordcount.jar"})
                .set("spark.driver.host", hostIp)
                .set("spark.driver.port", "8080")
                .set("spark.kubernetes.namespace", "default")
                .set("spark.kubernetes.container.image", "spark:3.3.2h.1")
                .set("spark.executor.cores", "2")
                .set("spark.executor.memory", "1g")
                .set("spark.kubernetes.authenticate.executor.serviceAccountName", "spark")
                .set("spark.kubernetes.dynamicAllocation.deleteGracePeriod", "20")
                .set("spark.cores.max", "4")
                .set("spark.executor.instances", "2");

        SparkSession spark = SparkSession.builder()
                .config(conf)
                .getOrCreate();

        byte[] byteArray = file.getBytes();
        String contents = new String(byteArray, StandardCharsets.UTF_8);
        Dataset<String> text = spark.createDataset(Arrays.asList(contents), Encoders.STRING());

        Dataset<String> wordsDataset = text.flatMap((FlatMapFunction<String, String>) line -> {
            List<String> words = new ArrayList<>();
            for (String word : line.split(" ")) {
                words.add(word);
            }
            return words.iterator();
        }, Encoders.STRING());

        // Count the number of occurrences of each word
        Dataset<Row> wordCounts = wordsDataset.groupBy("value")
                .agg(count("*").as("count"))
                .orderBy(desc("count"));

        // Convert the word count results to a List of Rows
        List<Row> wordCountsList = wordCounts.collectAsList();

        StringBuilder resultStringBuffer = new StringBuilder();

        // Build the final string representation
        for (Row row : wordCountsList) {
            resultStringBuffer.append(row.getString(0)).append(": ").append(row.getLong(1)).append("\n");
        }
        return ResponseEntity.ok(resultStringBuffer.toString());

    }

下面是我的maven 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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.8</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>wordcount</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>wordcount</name>
    <description>wordcount</description>
    <properties>
        <java.version>11</java.version>
        <spark.version>3.3.2</spark.version>
        <scala.version>2.12</scala.version>
    </properties>
    <dependencyManagement>
        <dependencies>
            <!--Spark java.lang.NoClassDefFoundError: org/codehaus/janino/InternalCompilerException-->
            <dependency>
                <groupId>org.codehaus.janino</groupId>
                <artifactId>commons-compiler</artifactId>
                <version>3.0.8</version>
            </dependency>
            <dependency>
                <groupId>org.codehaus.janino</groupId>
                <artifactId>janino</artifactId>
                <version>3.0.8</version>
            </dependency>
        </dependencies>
    </dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.codehaus.janino</groupId>
            <artifactId>commons-compiler</artifactId>
            <version>3.0.8</version>
        </dependency>
        <dependency>
            <groupId>org.codehaus.janino</groupId>
            <artifactId>janino</artifactId>
            <version>3.0.8</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency> <!-- Spark dependency -->
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_${scala.version}</artifactId>
            <version>${spark.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-kubernetes_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

下面是Dockerfile，我使用它来打包我的spring-boot应用程序，然后将其部署到minikube：

# Use an existing image as the base image
FROM openjdk:11-jdk

# Set the working directory
WORKDIR /app

# Copy the compiled JAR file to the image
COPY target/wordcount-0.0.1-SNAPSHOT.jar /app/wordcount.jar

RUN useradd -u 185 sparkuser

# Set the entrypoint command to run the JAR file
ENTRYPOINT ["java", "-jar", "wordcount.jar"]

对于spark.kubernetes.container.image，我使用本地Spark bin附带的Dockerfile（spark-3.3.2-bin-hadoop 3-与我的spring-boot应用程序使用的Spark版本相同）构建了一个docker镜像，并将其加载到minikube。
以下是我尝试过的一些事情，到目前为止没有运气：

按照这里的建议，使用setJars(new String[]{"/app/wordcount.jar"})与Spark共享我的应用程序的jar-这个绝对文件路径是我的应用程序的jar在我的驱动程序映像上的位置
使用maven-shade-plugin作为建议here来改变我的应用程序的jar分发其依赖项的方式-这导致了我的驱动程序pod上的ClassNotFoundException: SparkSession异常。
重构我的控制器代码，不使用lambda函数（没有区别）：

public static class SplitLine implements FlatMapFunction<String, String> {
   @Override
   public Iterator<String> call(String line) throws Exception {
       List<String> words = new ArrayList<>();
       for (String word : line.split(" ")) {
           words.add(word);
       }
       return words.iterator();
   }

...

Dataset<String> wordsDataset = text.flatMap(new SplitLine(), Encoders.STRING());

任何关于我的设置的提示或建议，或者关于如何重构我的代码以使其与现有设置一起工作的建议，都将不胜感激。

kubernetes Sping Boot Spark on K8S(Minikube)：无法分配java.lang.invoke.SerializedLambda的示例

1条答案

相关问题

热门标签

最新问答