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

@Component
@RequiredArgsConstructor
public class SomeService implements FlatMapFunction<T, K> {

  private final ObjectMapper mapper;

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

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)

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

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;
    }
}