本文探讨了在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<T, K> {
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。然而，这种方式的缺点显而易见：
- 失去Spring的集中管理： 手动创建的ObjectMapper实例不会继承Spring应用上下文中为ObjectMapper配置的任何全局设置（如日期格式、模块注册、序列化/反序列化特性等）。这可能导致行为不一致。
- 配置分散： 如果有多个Spark任务需要ObjectMapper，每个任务都需要重复配置，增加了维护成本和出错风险。
- 资源浪费： 每次实例化都可能创建新的资源，而不是重用Spring管理的单例。
因此，我们需要一个更优雅、更符合Spring DI原则的解决方案。
核心解决方案：自定义可序列化的HandlerInstantiator
解决此问题的关键在于替换ObjectMapper中不可序列化的SpringHandlerInstantiator。我们可以创建一个自定义的、可序列化的HandlerInstantiator实现，并在Spring配置中将它设置给ObjectMapper Bean。
com.fasterxml.jackson.databind.cfg.HandlerInstantiator是Jackson提供的一个接口，它允许我们自定义如何实例化各种处理器（如JsonSerializer、JsonDeserializer、KeyDeserializer等）。SpringHandlerInstantiator正是利用此机制，使得Jackson能够使用Spring容器管理的自定义序列化/反序列化器。
为了解决序列化问题，我们需要：
- 创建一个可序列化的HandlerInstantiator实现：这个实现（例如命名为MyHandlerInstantiator）必须是可序列化的，并且如果需要，可以模拟SpringHandlerInstantiator的部分功能，但要避免持有不可序列化的Spring上下文引用。在许多简单场景中，如果您的应用没有复杂的自定义Jackson模块或处理器需要Spring上下文来实例化，一个简单的、不依赖Spring上下文的HandlerInstantiator就足够了。如果需要，MyHandlerInstantiator可以简单地通过反射或Class.forName()来创建处理器实例。
- 在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集群中的工作节点执行。
注意事项：
- MyHandlerInstantiator的复杂性： 如果您的Jackson配置非常依赖于Spring上下文来实例化自定义序列化器、反序列化器或类型解析器，那么MyHandlerInstantiator的实现可能会变得复杂。您可能需要仔细考虑如何将这些Spring管理的依赖转换为可序列化的形式，或者在Spark任务执行时重新构建它们。在大多数情况下，简单的应用可能只需要一个不依赖Spring上下文的HandlerInstantiator。
- Jackson模块的注册： 确保在自定义objectMapper() Bean方法中注册所有必需的Jackson模块（例如JavaTimeModule用于处理Java 8日期时间API），以便ObjectMapper具备完整的序列化/反序列化能力。
- 兼容性： 这种方法在Spring Boot和Spark的兼容性方面表现良好，它允许您在保持Spring DI优势的同时，满足Spark的序列化要求。
总结
在Spring Boot与Spark集成时，ObjectMapper的序列化问题是一个常见挑战，其核心在于SpringHandlerInstantiator的不可序列化性。通过实现一个自定义的、可序列化的HandlerInstantiator，并将其配置到Spring管理的ObjectMapper Bean中，我们可以有效地解决Task not serializable错误，从而确保Spark任务能够顺利地在分布式环境中执行，同时保留Spring依赖注入的便利性和ObjectMapper的统一配置。这种方法提供了一个健壮且符合最佳实践的解决方案，避免了手动实例化ObjectMapper所带来的配置分散和一致性问题。
