解决Spark与Spring Boot中ObjectMapper的序列化问题

本文探讨了在spring boot应用中将objectmapper注入spark任务时遇到的`task not serializable`问题。核心原因是`springhandlerinstantiator`不可序列化。教程提供了一种解决方案，通过配置一个自定义的、可序列化的`handlerinstantiator`来确保`objectmapper`及其依赖能够被spark正确序列化，从而实现在分布式环境中无缝使用spring管理的`objectmapper`。

Spark分布式任务中的序列化挑战

在Apache Spark等分布式计算框架中，为了在集群的不同节点上执行任务，所有涉及到的闭包、函数以及其依赖的对象都必须是可序列化的。当我们在Spring Boot应用中开发Spark任务（例如实现FlatMapFunction或MapFunction接口的组件）时，如果这些组件依赖于Spring容器管理的Bean，尤其是像ObjectMapper这样的复杂对象，就可能遇到序列化问题。

考虑以下Spring Boot组件，它作为Spark的FlatMapFunction使用，并依赖于通过Spring依赖注入（DI）的ObjectMapper：

@Component
@RequiredArgsConstructor
public class SomeService implements FlatMapFunction {

  private final ObjectMapper mapper;

  // ... 其他业务逻辑和apply方法实现
}

当Spark尝试序列化SomeService实例并将其分发到工作节点时，如果ObjectMapper或其内部依赖不可序列化，就会抛出org.apache.spark.SparkException: Task not serializable异常。

问题根源：SpringHandlerInstantiator的不可序列化性

通过分析Spark抛出的序列化堆栈信息，我们可以定位到问题的具体原因：

Caused by: java.io.NotSerializableException: org.springframework.http.converter.json.SpringHandlerInstantiator
Serialization stack:
- object not serializable (class: org.springframework.http.converter.json.SpringHandlerInstantiator, value: org.springframework.http.converter.json.SpringHandlerInstantiator@6e4912db)
- field (class: com.fasterxml.jackson.databind.cfg.BaseSettings, name: _handlerInstantiator, type: class com.fasterxml.jackson.databind.cfg.HandlerInstantiator)
- object (class com.fasterxml.jackson.databind.cfg.BaseSettings, com.fasterxml.jackson.databind.cfg.BaseSettings@155616d8)
- field (class: com.fasterxml.jackson.databind.cfg.MapperConfig, name: _base, type: class com.fasterxml.jackson.databind.cfg.BaseSettings)
- object (class com.fasterxml.jackson.databind.DeserializationConfig, com.fasterxml.jackson.databind.DeserializationConfig@66e72ca2)
- field (class: com.fasterxml.jackson.databind.ObjectMapper, name: _deserializationConfig, type: class com.fasterxml.jackson.databind.DeserializationConfig)
- object (class com.fasterxml.jackson.databind.ObjectMapper, com.fasterxml.jackson.databind.ObjectMapper@433ef204)
- field (class: com.smth.SomeService, name: mapper, type: class com.fasterxml.jackson.databind.ObjectMapper)

从堆栈中可以看到，ObjectMapper内部的_deserializationConfig字段依赖于MapperConfig，而MapperConfig又依赖于BaseSettings，最终BaseSettings持有的_handlerInstantiator字段是一个org.springframework.http.converter.json.SpringHandlerInstantiator实例。这个SpringHandlerInstantiator是Spring框架为了集成Jackson而提供的，它通常会持有对Spring应用上下文的引用，而Spring应用上下文本身是不可序列化的。因此，当Spark尝试序列化ObjectMapper时，由于其深层依赖SpringHandlerInstantiator不可序列化，整个SomeService实例的序列化过程就会失败。

临时解决方案及其局限性

一种常见的临时解决方案是在SomeService的构造函数中手动实例化ObjectMapper：

public SomeService() {
  this.mapper = new ObjectMapper();
}

这种方法确实可以解决序列化问题，因为它不再依赖于Spring注入的ObjectMapper，从而避免了SpringHandlerInstantiator。然而，这种方式的缺点显而易见：

1. 失去Spring的集中管理： 手动创建的ObjectMapper实例不会继承Spring应用上下文中为ObjectMapper配置的任何全局设置（如日期格式、模块注册、序列化/反序列化特性等）。这可能导致行为不一致。
2. 配置分散： 如果有多个Spark任务需要ObjectMapper，每个任务都需要重复配置，增加了维护成本和出错风险。
3. 资源浪费： 每次实例化都可能创建新的资源，而不是重用Spring管理的单例。

因此，我们需要一个更优雅、更符合Spring DI原则的解决方案。

核心解决方案：自定义可序列化的HandlerInstantiator

解决此问题的关键在于替换ObjectMapper中不可序列化的SpringHandlerInstantiator。我们可以创建一个自定义的、可序列化的HandlerInstantiator实现，并在Spring配置中将它设置给ObjectMapper Bean。

Pascal基础教程 Pascal入门必备基础教程 CHM版

无论做任何事情，都要有一定的方式方法与处理步骤。计算机程序设计比日常生活中的事务处理更具有严谨性、规范性、可行性。为了使计算机有效地解决某些问题，须将处理步骤编排好，用计算机语言组成“序列”，让计算机自动识别并执行这个用计算机语言组成的“序列”，完成预定的任务。将处理问题的步骤编排好，用计算机语言组成序列，也就是常说的编写程序。在Pascal语言中，执行每条语句都是由计算机完成相应的操作。编写Pascal程序，是利用Pasca

下载

com.fasterxml.jackson.databind.cfg.HandlerInstantiator是Jackson提供的一个接口，它允许我们自定义如何实例化各种处理器（如JsonSerializer、JsonDeserializer、KeyDeserializer等）。SpringHandlerInstantiator正是利用此机制，使得Jackson能够使用Spring容器管理的自定义序列化/反序列化器。

为了解决序列化问题，我们需要：

1. 创建一个可序列化的HandlerInstantiator实现：这个实现（例如命名为MyHandlerInstantiator）必须是可序列化的，并且如果需要，可以模拟SpringHandlerInstantiator的部分功能，但要避免持有不可序列化的Spring上下文引用。在许多简单场景中，如果您的应用没有复杂的自定义Jackson模块或处理器需要Spring上下文来实例化，一个简单的、不依赖Spring上下文的HandlerInstantiator就足够了。如果需要，MyHandlerInstantiator可以简单地通过反射或Class.forName()来创建处理器实例。
2. 在Spring配置中注册带有自定义HandlerInstantiator的ObjectMapper Bean。

示例代码：配置自定义ObjectMapper Bean

下面是一个Spring配置示例，展示如何注册一个使用自定义HandlerInstantiator的ObjectMapper Bean：

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.cfg.HandlerInstantiator;
import com.fasterxml.jackson.databind.deser.ValueInstantiator;
import com.fasterxml.jackson.databind.introspect.Annotated;
import com.fasterxml.jackson.databind.introspect.AnnotatedClass;
import com.fasterxml.jackson.databind.introspect.AnnotatedMember;
import com.fasterxml.jackson.databind.JsonDeserializer;
import com.fasterxml.jackson.databind.JsonSerializer;
import com.fasterxml.jackson.databind.KeyDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.Serializable;

@Configuration
public class JacksonConfig {

    /**
     * 自定义的可序列化HandlerInstantiator。
     * 在大多数简单场景下，如果您的应用不依赖Spring上下文来实例化自定义Jackson模块，
     * 它可以是一个简单的实现。如果需要，可以扩展SpringHandlerInstantiator并移除不可序列化的部分，
     * 或者根据具体需求进行定制。
     */
    public static class MyHandlerInstantiator extends HandlerInstantiator implements Serializable {

        private static final long serialVersionUID = 1L; // 确保可序列化

        @Override
        public JsonDeserializer deserializerInstance(
                com.fasterxml.jackson.databind.DeserializationConfig config,
                Annotated annotated,
                Class deserClass) {
            // 这里可以添加逻辑来从Spring上下文获取或手动创建反序列化器实例
            // 对于不依赖Spring上下文的简单情况，直接通过反射创建即可
            try {
                return (JsonDeserializer) deserClass.getDeclaredConstructor().newInstance();
            } catch (Exception e) {
                throw new RuntimeException("Failed to instantiate deserializer: " + deserClass.getName(), e);
            }
        }

        @Override
        public KeyDeserializer keyDeserializerInstance(
                com.fasterxml.jackson.databind.DeserializationConfig config,
                Annotated annotated,
                Class keyDeserClass) {
            try {
                return (KeyDeserializer) keyDeserClass.getDeclaredConstructor().newInstance();
            } catch (Exception e) {
                throw new RuntimeException("Failed to instantiate key deserializer: " + keyDeserClass.getName(), e);
            }
        }

        @Override
        public JsonSerializer serializerInstance(
                com.fasterxml.jackson.databind.SerializationConfig config,
                Annotated annotated,
                Class serClass) {
            try {
                return (JsonSerializer) serClass.getDeclaredConstructor().newInstance();
            } catch (Exception e) {
                throw new RuntimeException("Failed to instantiate serializer: " + serClass.getName(), e);
            }
        }

        @Override
        public ValueInstantiator valueInstantiatorInstance(
                com.fasterxml.jackson.databind.DeserializationConfig config,
                AnnotatedClass annotatedClass,
                Class valueInstantiatorClass) {
            try {
                return (ValueInstantiator) valueInstantiatorClass.getDeclaredConstructor().newInstance();
            } catch (Exception e) {
                throw new RuntimeException("Failed to instantiate value instantiator: " + valueInstantiatorClass.getName(), e);
            }
        }

        @Override
        public Object resolverIdGeneratorInstance(
                com.fasterxml.jackson.databind.DeserializationConfig config,
                Annotated annotated,
                Class implClass) {
            try {
                return implClass.getDeclaredConstructor().newInstance();
            } catch (Exception e) {
                throw new RuntimeException("Failed to instantiate resolver id generator: " + implClass.getName(), e);
            }
        }
    }

    @Bean
    public ObjectMapper objectMapper() {
        ObjectMapper mapper = new ObjectMapper();
        // 设置自定义的可序列化HandlerInstantiator
        mapper.setHandlerInstantiator(new MyHandlerInstantiator());
        // 可以继续配置ObjectMapper的其他属性，例如注册模块等
        // mapper.registerModule(new JavaTimeModule());
        // mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
        return mapper;
    }
}

说明：

• MyHandlerInstantiator类必须实现Serializable接口。
• 在MyHandlerInstantiator的实现中，我们通过反射来创建处理器实例。这种方法适用于处理器本身没有复杂依赖，或者其依赖可以通过其他可序列化方式提供的场景。
• 如果您的应用中有复杂的自定义序列化器/反序列化器，并且这些处理器本身依赖于Spring容器中的其他Bean，那么MyHandlerInstantiator的实现就需要更复杂一些，例如在实例化时通过某种机制（如静态工厂方法或在构造函数中传递可序列化的配置）来获取这些依赖，而不是直接引用不可序列化的ApplicationContext。在许多情况下，简单的反射实例化已经足够。
• JacksonConfig类中的objectMapper方法会覆盖Spring Boot自动配置的ObjectMapper Bean，确保Spark任务获取到的是我们自定义配置的ObjectMapper。

工作原理与注意事项

通过上述配置，我们用一个可序列化的MyHandlerInstantiator替换了ObjectMapper中默认的SpringHandlerInstantiator。当Spark尝试序列化SomeService时，它会沿着依赖链找到ObjectMapper，进而找到MyHandlerInstantiator。由于MyHandlerInstantiator是可序列化的，整个SomeService实例就可以被成功序列化并分发到Spark集群中的工作节点执行。

注意事项：

1. MyHandlerInstantiator的复杂性： 如果您的Jackson配置非常依赖于Spring上下文来实例化自定义序列化器、反序列化器或类型解析器，那么MyHandlerInstantiator的实现可能会变得复杂。您可能需要仔细考虑如何将这些Spring管理的依赖转换为可序列化的形式，或者在Spark任务执行时重新构建它们。在大多数情况下，简单的应用可能只需要一个不依赖Spring上下文的HandlerInstantiator。
2. Jackson模块的注册： 确保在自定义objectMapper() Bean方法中注册所有必需的Jackson模块（例如JavaTimeModule用于处理Java 8日期时间API），以便ObjectMapper具备完整的序列化/反序列化能力。
3. 兼容性： 这种方法在Spring Boot和Spark的兼容性方面表现良好，它允许您在保持Spring DI优势的同时，满足Spark的序列化要求。

总结

在Spring Boot与Spark集成时，ObjectMapper的序列化问题是一个常见挑战，其核心在于SpringHandlerInstantiator的不可序列化性。通过实现一个自定义的、可序列化的HandlerInstantiator，并将其配置到Spring管理的ObjectMapper Bean中，我们可以有效地解决Task not serializable错误，从而确保Spark任务能够顺利地在分布式环境中执行，同时保留Spring依赖注入的便利性和ObjectMapper的统一配置。这种方法提供了一个健壮且符合最佳实践的解决方案，避免了手动实例化ObjectMapper所带来的配置分散和一致性问题。