java - 如何修复 wordCountProgram 的 : java. io.NotSerializedException : org. apache.spark.api.java.JavaSparkContext

Question

我尝试在 springboot 中编写一个简单的休息 Controller ，它接受一个文件和一个字符串，并在该文件中搜索该字符串被找到的次数，基本上是单词搜索和计数。

我的电脑上运行着 master 和 runner，并且 springboot 应用程序连接没有任何问题。我尝试在没有“专用”spark master 的情况下使用，它工作正常，但在我将其连接到我想要的问题后，问题开始出现。

我也尝试过使用 lambda 表达式，但这给我带来了其他问题，所以我试图让它变得更简单。

Spark 配置:

@Configuration
public class SparkConfig {

    @Value("${spark.app.name}")
    private String appname;

    @Value("${spark.master}")
    private String masterUri;

    @Bean
    public SparkConf conf(){
        System.out.println(appname + " " + masterUri);
        return new SparkConf()
                .setAppName(appname)
                .setMaster(masterUri);
    }

    @Bean
    public JavaSparkContext sc(){
        return new JavaSparkContext(conf());
    }

端点:

  @RequestMapping(value = "/file-word-count",method = RequestMethod.POST)
    public String fileWordCount(@RequestParam("file") MultipartFile file, @RequestParam String word)  {
       return wordCountService.countFileWords(file,word);
    }

服务

@Service
@Component
public class WordCountService implements Serializable {

    @Autowired
    JavaSparkContext sc;


    public String countFileWords(MultipartFile file, String word)  {
        String result = null;

        try {
            JavaRDD<String> textFile = sc.textFile(convertMultiToFile(file).getPath());
            JavaRDD<String> words = textFile.filter(new Function<String, Boolean>() {
                @Override
                public Boolean call(String s) throws Exception {
                    return s.contains(word);
                }
            });

            result = String.valueOf(words.count());

        }
        catch(IOException e){
                System.out.println(e.getMessage());
            }

        return result;
        }



    public File convertMultiToFile(MultipartFile mFile) throws IOException {
        File file = new File("temp","tempTextFile");
        FileUtils.writeByteArrayToFile(file,mFile.getBytes());
            return file;
    }
}

控制台输出:

2019-06-25 22:11:08.462 ERROR 5960 --- [nio-9090-exec-1] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.apache.spark.SparkException: Task not serializable] with root cause

java.io.NotSerializableException: org.apache.spark.api.java.JavaSparkContext
Serialization stack:
    - object not serializable (class: org.apache.spark.api.java.JavaSparkContext, value: org.apache.spark.api.java.JavaSparkContext@47c997cb)
    - field (class: com.licenta.service.WordCountService, name: sc, type: class org.apache.spark.api.java.JavaSparkContext)
    - object (class com.licenta.service.WordCountService, com.licenta.service.WordCountService@4152741c)
    - field (class: com.licenta.service.WordCountService$1, name: this$0, type: class com.licenta.service.WordCountService)
    - object (class com.licenta.service.WordCountService$1, com.licenta.service.WordCountService$1@22ae6915)
    - field (class: org.apache.spark.api.java.JavaRDD$$anonfun$filter$1, name: f$1, type: interface org.apache.spark.api.java.function.Function)
    - object (class org.apache.spark.api.java.JavaRDD$$anonfun$filter$1, <function1>)
    at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:400) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:393) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:162) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.SparkContext.clean(SparkContext.scala:2326) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:388) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:387) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:363) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.rdd.RDD.filter(RDD.scala:387) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at org.apache.spark.api.java.JavaRDD.filter(JavaRDD.scala:78) ~[spark-core_2.11-2.4.3.jar:2.4.3]
    at com.licenta.service.WordCountService.countFileWords(WordCountService.java:35) ~[classes/:na]
    at com.licenta.controller.CountController.fileWordCount(CountController.java:29) ~[classes/:na]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_161]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_161]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_161]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_161]
    at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:189) ~[spring-web-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:138) ~[spring-web-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:102) ~[spring-webmvc-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:892) ~[spring-webmvc-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:797) ~[spring-webmvc-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87) ~[spring-webmvc-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1038) ~[spring-webmvc-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:942) ~[spring-webmvc-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1005) ~[spring-webmvc-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.servlet.FrameworkServlet.doPost(FrameworkServlet.java:908) ~[spring-webmvc-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at javax.servlet.http.HttpServlet.service(HttpServlet.java:660) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:882) ~[spring-webmvc-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at javax.servlet.http.HttpServlet.service(HttpServlet.java:741) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:231) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:53) ~[tomcat-embed-websocket-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:99) ~[spring-web-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107) ~[spring-web-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:92) ~[spring-web-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107) ~[spring-web-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.springframework.web.filter.HiddenHttpMethodFilter.doFilterInternal(HiddenHttpMethodFilter.java:93) ~[spring-web-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107) ~[spring-web-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:200) ~[spring-web-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107) ~[spring-web-5.1.6.RELEASE.jar:5.1.6.RELEASE]
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:200) ~[tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:96) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:490) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:139) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:92) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:74) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:343) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:408) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:66) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:834) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1415) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_161]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_161]
    at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61) [tomcat-embed-core-9.0.17.jar:9.0.17]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_161]

2019-06-25 22:41:00.704  INFO 5960 --- [er-event-loop-8] o.apache.spark.storage.BlockManagerInfo  : Removed broadcast_0_piece0 on DESKTOP-TF9C4EG:55462 in memory (size: 20.4 KB, free: 898.5 MB)

我希望得到一个包含单词出现次数的字符串。希望我能清楚地解释，如果没有，请告诉我，我会尽力回答。

Answer 1

问题是new Function<String, Boolean>() ，它是一个匿名类，并且引用 WordCountService并传递至JavaSparkContext 。为了避免这种情况，您可以将其设为静态嵌套类。

static class WordCounter implements Function<String, Boolean>, Serializable {
       private final String word;
       public WordCounter(String word){
           this.word = word;
       }

       @Override
       public Boolean call(String s) throws Exception {
                return s.contains(word);
       }
 }

并使用它

  JavaRDD<String> words = textFile.filter(new WordCounter(word));