🔀 S&P 선박제원정보 증분데이터 수집JOB

🔀 S&P 선박제원정보 증분데이터 수집JOB (Squash)

pom.xml

@@ -56,7 +56,6 @@
             <groupId>org.postgresql</groupId>
             <artifactId>postgresql</artifactId>
             <version>${postgresql.version}</version>
-            <scope>runtime</scope>
         </dependency>
 
         <!-- Spring Boot Starter Thymeleaf (for Web GUI) -->

src/main/java/com/snp/batch/common/batch/repository/BaseJdbcRepository.java

@@ -39,6 +39,9 @@ public abstract class BaseJdbcRepository<T, ID> {
     protected String getIdColumnName() {
         return "id";
     }
+    protected String getIdColumnName(String customId) {
+        return customId;
+    }
 
     /**
      * RowMapper 반환 (하위 클래스에서 구현)

src/main/java/com/snp/batch/common/util/JsonChangeDetector.java

@@ -0,0 +1,179 @@
+package com.snp.batch.common.util;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.security.MessageDigest;
+import java.util.*;
+
+public class JsonChangeDetector {
+
+    // Map으로 변환 시 사용할 ObjectMapper (표준 Mapper 사용)
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+
+    // 해시 비교에서 제외할 필드 목록 (DataSetVersion 등)
+    // 이 목록은 모든 JSON 계층에 걸쳐 적용됩니다.
+    private static final java.util.Set<String> EXCLUDE_KEYS =
+            java.util.Set.of("DataSetVersion", "APSStatus", "LastUpdateDate", "LastUpdateDateTime");
+
+    private static final Map<String, String> LIST_SORT_KEYS = Map.of(
+            // List 필드명         // 정렬 기준 키
+            "OwnerHistory"              ,"Sequence",     // OwnerHistory는 Sequence를 기준으로 정렬
+            "SurveyDatesHistoryUnique"  , "SurveyDate"      // SurveyDatesHistoryUnique는 SurveyDate를 기준으로 정렬
+            // 추가적인 List/Array 필드가 있다면 여기에 추가
+    );
+
+    // =========================================================================
+    // 1. JSON 문자열을 정렬 및 필터링된 Map으로 변환하는 핵심 로직
+    // =========================================================================
+    /**
+     * JSON 문자열을 Map으로 변환하고, 특정 키를 제거하며, 키 순서가 정렬된 상태로 만듭니다.
+     * @param jsonString API 응답 또는 DB에서 읽은 JSON 문자열
+     * @return 필터링되고 정렬된 Map 객체
+     */
+    public static Map<String, Object> jsonToSortedFilteredMap(String jsonString) {
+        if (jsonString == null || jsonString.trim().isEmpty()) {
+            return Collections.emptyMap();
+        }
+
+        try {
+            // 1. Map<String, Object>으로 1차 변환합니다. (순서 보장 안됨)
+            Map<String, Object> rawMap = MAPPER.readValue(jsonString,
+                    new com.fasterxml.jackson.core.type.TypeReference<Map<String, Object>>() {});
+
+            // 2. 재귀 함수를 호출하여 키를 제거하고 TreeMap(키 순서 정렬)으로 깊은 복사합니다.
+            return deepFilterAndSort(rawMap);
+
+        } catch (Exception e) {
+            System.err.println("Error converting JSON to filtered Map: " + e.getMessage());
+            // 예외 발생 시 빈 Map 반환
+            return Collections.emptyMap();
+        }
+    }
+
+    /**
+     * Map을 재귀적으로 탐색하며 제외 키를 제거하고 TreeMap(알파벳 순서)으로 변환합니다.
+     */
+    private static Map<String, Object> deepFilterAndSort(Map<String, Object> rawMap) {
+        // Map을 TreeMap으로 생성하여 키 순서를 알파벳 순으로 강제 정렬합니다.
+        Map<String, Object> sortedMap = new TreeMap<>();
+
+        for (Map.Entry<String, Object> entry : rawMap.entrySet()) {
+            String key = entry.getKey();
+            Object value = entry.getValue();
+
+            // 🔑 1. 제외할 키 값인지 확인
+            if (EXCLUDE_KEYS.contains(key)) {
+                continue; // 제외
+            }
+
+            // 2. 값의 타입에 따라 재귀 처리
+            if (value instanceof Map) {
+                // 재귀 호출: 하위 Map을 필터링하고 정렬
+                @SuppressWarnings("unchecked")
+                Map<String, Object> subMap = (Map<String, Object>) value;
+                sortedMap.put(key, deepFilterAndSort(subMap));
+            } else if (value instanceof List) {
+                // List 처리: List 내부의 Map 요소만 재귀 호출
+                @SuppressWarnings("unchecked")
+                List<Object> rawList = (List<Object>) value;
+                List<Object> filteredList = new ArrayList<>();
+
+                // 1. List 내부의 Map 요소들을 재귀적으로 필터링/정렬하여 filteredList에 추가
+                for (Object item : rawList) {
+                    if (item instanceof Map) {
+                        @SuppressWarnings("unchecked")
+                        Map<String, Object> itemMap = (Map<String, Object>) item;
+                        // List의 요소인 Map도 필터링하고 정렬 (Map의 필드 순서 정렬)
+                        filteredList.add(deepFilterAndSort(itemMap));
+                    } else {
+                        filteredList.add(item);
+                    }
+                }
+
+                // 2. 🔑 List 필드명에 따른 순서 정렬 로직 (추가된 핵심 로직)
+                String listFieldName = entry.getKey();
+                String sortKey = LIST_SORT_KEYS.get(listFieldName);
+
+                if (sortKey != null && !filteredList.isEmpty() && filteredList.get(0) instanceof Map) {
+                    // Map 요소를 가진 리스트인 경우에만 정렬 실행
+                    try {
+                        // 정렬 기준 키를 사용하여 Comparator를 생성
+                        Collections.sort(filteredList, new Comparator<Object>() {
+                            @Override
+                            @SuppressWarnings("unchecked")
+                            public int compare(Object o1, Object o2) {
+                                Map<String, Object> map1 = (Map<String, Object>) o1;
+                                Map<String, Object> map2 = (Map<String, Object>) o2;
+
+                                // 정렬 기준 키(sortKey)의 값을 가져와 비교
+                                Object key1 = map1.get(sortKey);
+                                Object key2 = map2.get(sortKey);
+
+                                if (key1 == null || key2 == null) {
+                                    // 키 값이 null인 경우, Map의 전체 문자열로 비교 (안전장치)
+                                    return map1.toString().compareTo(map2.toString());
+                                }
+
+                                // String 타입으로 변환하여 비교 (Date, Number 타입도 대부분 String으로 처리 가능)
+                                return key1.toString().compareTo(key2.toString());
+                            }
+                        });
+                    } catch (Exception e) {
+                        System.err.println("List sort failed for key " + listFieldName + ": " + e.getMessage());
+                        // 정렬 실패 시 원래 순서 유지
+                    }
+                }
+                sortedMap.put(key, filteredList);
+            } else {
+                // String, Number 등 기본 타입은 그대로 추가
+                sortedMap.put(key, value);
+            }
+        }
+        return sortedMap;
+    }
+
+
+    // =========================================================================
+    // 2. 해시 생성 로직
+    // =========================================================================
+
+    /**
+     * 필터링되고 정렬된 Map의 문자열 표현을 기반으로 SHA-256 해시를 생성합니다.
+     */
+    public static String getSha256HashFromMap(Map<String, Object> sortedMap) {
+        // 1. Map을 String으로 변환: TreeMap 덕분에 toString() 결과가 항상 동일한 순서를 가집니다.
+        String mapString = sortedMap.toString();
+
+        try {
+            MessageDigest digest = MessageDigest.getInstance("SHA-256");
+            byte[] hash = digest.digest(mapString.getBytes("UTF-8"));
+
+            // 바이트 배열을 16진수 문자열로 변환
+            StringBuilder hexString = new StringBuilder();
+            for (byte b : hash) {
+                String hex = Integer.toHexString(0xff & b);
+                if (hex.length() == 1) hexString.append('0');
+                hexString.append(hex);
+            }
+            return hexString.toString();
+        } catch (Exception e) {
+            System.err.println("Error generating hash: " + e.getMessage());
+            return "HASH_ERROR";
+        }
+    }
+
+
+    // =========================================================================
+    // 3. 해시값 비교 로직
+    // =========================================================================
+    public static boolean isChanged(String previousHash, String currentHash) {
+        // DB 해시가 null인 경우 (첫 Insert)는 변경된 것으로 간주
+        if (previousHash == null || previousHash.isEmpty()) {
+            return true;
+        }
+
+        // 해시값이 다르면 변경된 것으로 간주
+        return !Objects.equals(previousHash, currentHash);
+    }
+
+}

src/main/java/com/snp/batch/global/config/SwaggerConfig.java

@@ -38,6 +38,9 @@ public class SwaggerConfig {
                         new Server()
                                 .url("http://localhost:" + serverPort)
                                 .description("로컬 개발 서버"),
+                        new Server()
+                                .url("http://211.208.115.83:" + serverPort)
+                                .description("중계 서버"),
                         new Server()
                                 .url("https://api.snp-batch.com")
                                 .description("운영 서버 (예시)")

src/main/java/com/snp/batch/jobs/shipdetail/batch/config/ShipDetailImportJobConfig.java

@@ -1,8 +1,8 @@
 package com.snp.batch.jobs.shipdetail.batch.config;
 
 import com.snp.batch.common.batch.config.BaseJobConfig;
-import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailDto;
-import com.snp.batch.jobs.shipdetail.batch.entity.ShipDetailEntity;
+import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailComparisonData; // Reader 출력 타입
+import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailUpdate; // Processor 출력 타입
 import com.snp.batch.jobs.shipdetail.batch.processor.ShipDetailDataProcessor;
 import com.snp.batch.jobs.shipdetail.batch.reader.ShipDetailDataReader;
 import com.snp.batch.jobs.shipdetail.batch.writer.ShipDetailDataWriter;
@@ -19,6 +19,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.web.reactive.function.client.WebClient;
+import com.fasterxml.jackson.databind.ObjectMapper; // ObjectMapper 추가
 
 /**
  * 선박 상세 정보 Import Job Config
@@ -27,6 +28,7 @@ import org.springframework.web.reactive.function.client.WebClient;
  * - ship_data 테이블에서 IMO 번호 조회
  * - IMO 번호를 100개씩 배치로 분할
  * - Maritime API GetShipsByIHSLRorIMONumbers 호출
+ *         TODO : GetShipsByIHSLRorIMONumbersAll 호출로 변경
  * - 선박 상세 정보를 ship_detail 테이블에 저장 (UPSERT)
  *
  * 데이터 흐름:
@@ -37,31 +39,35 @@ import org.springframework.web.reactive.function.client.WebClient;
  * ShipDetailDataWriter
  *   ↓ (ship_detail 테이블)
  */
+/**
+ * 선박 상세 정보 Import Job Config
+ * I: ShipDetailComparisonData (Reader 출력)
+ * O: ShipDetailUpdate (Processor 출력)
+ */
 @Slf4j
 @Configuration
-public class ShipDetailImportJobConfig extends BaseJobConfig<ShipDetailDto, ShipDetailEntity> {
+public class ShipDetailImportJobConfig extends BaseJobConfig<ShipDetailComparisonData, ShipDetailUpdate> {
 
     private final ShipDetailDataProcessor shipDetailDataProcessor;
     private final ShipDetailDataWriter shipDetailDataWriter;
     private final JdbcTemplate jdbcTemplate;
     private final WebClient maritimeApiWebClient;
+    private final ObjectMapper objectMapper; // ObjectMapper 주입 추가
 
-    /**
-     * 생성자 주입
-     * maritimeApiWebClient: MaritimeApiWebClientConfig에서 등록한 Bean 주입 (ShipImportJob과 동일)
-     */
     public ShipDetailImportJobConfig(
             JobRepository jobRepository,
             PlatformTransactionManager transactionManager,
             ShipDetailDataProcessor shipDetailDataProcessor,
             ShipDetailDataWriter shipDetailDataWriter,
             JdbcTemplate jdbcTemplate,
-            @Qualifier("maritimeApiWebClient") WebClient maritimeApiWebClient) {
+            @Qualifier("maritimeApiWebClient") WebClient maritimeApiWebClient,
+            ObjectMapper objectMapper) { // ObjectMapper 주입 추가
         super(jobRepository, transactionManager);
         this.shipDetailDataProcessor = shipDetailDataProcessor;
         this.shipDetailDataWriter = shipDetailDataWriter;
         this.jdbcTemplate = jdbcTemplate;
         this.maritimeApiWebClient = maritimeApiWebClient;
+        this.objectMapper = objectMapper; // ObjectMapper 초기화
     }
 
     @Override
@@ -75,23 +81,24 @@ public class ShipDetailImportJobConfig extends BaseJobConfig<ShipDetailDto, Ship
     }
 
     @Override
-    protected ItemReader<ShipDetailDto> createReader() {
-        return new ShipDetailDataReader(maritimeApiWebClient, jdbcTemplate);
+    protected ItemReader<ShipDetailComparisonData> createReader() { // 타입 변경
+        // Reader 생성자 수정: ObjectMapper를 전달합니다.
+        return new ShipDetailDataReader(maritimeApiWebClient, jdbcTemplate, objectMapper);
     }
 
     @Override
-    protected ItemProcessor<ShipDetailDto, ShipDetailEntity> createProcessor() {
+    protected ItemProcessor<ShipDetailComparisonData, ShipDetailUpdate> createProcessor() {
         return shipDetailDataProcessor;
     }
 
     @Override
-    protected ItemWriter<ShipDetailEntity> createWriter() {
+    protected ItemWriter<ShipDetailUpdate> createWriter() { // 타입 변경
         return shipDetailDataWriter;
     }
 
     @Override
     protected int getChunkSize() {
-        return 100; // API에서 100개씩 가져오므로 chunk도 100으로 설정
+        return 50; // API에서 100개씩 가져오므로 chunk도 100으로 설정
     }
 
     @Bean(name = "shipDetailImportJob")

src/main/java/com/snp/batch/jobs/shipdetail/batch/dto/OwnerHistoryDto.java

@@ -0,0 +1,52 @@
+package com.snp.batch.jobs.shipdetail.batch.dto;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class OwnerHistoryDto {
+    /**
+     * 회사 상태
+     * API: CompanyStatus
+     */
+    @JsonProperty("CompanyStatus")
+    private String CompanyStatus;
+    /**
+     * 효력 일자
+     * API: EffectiveDate
+     */
+    @JsonProperty("EffectiveDate")
+    private String EffectiveDate;
+    /**
+     * IMO 번호
+     * API: LRNO
+     */
+    @JsonProperty("LRNO")
+    private String LRNO;
+    /**
+     * 소유주
+     * API: Owner
+     */
+    @JsonProperty("Owner")
+    private String Owner;
+    /**
+     * 소유주 코드
+     * API: OwnerCode
+     */
+    @JsonProperty("OwnerCode")
+    private String OwnerCode;
+    /**
+     * 시퀀스
+     * API: Sequence
+     */
+    @JsonProperty("Sequence")
+    private String Sequence;
+}

src/main/java/com/snp/batch/jobs/shipdetail/batch/dto/ShipDetailApiResponse.java

@@ -16,6 +16,16 @@ import java.util.List;
  *   "shipCount": 5,
  *   "Ships": [...]
  * }
+ *
+ * Maritime API GetShipsByIHSLRorIMONumbersAll 응답 래퍼
+ *
+ * API 응답 구조:
+ * {
+ *     "shipCount": 5,
+ *     "ShipResult": [...],
+ *     "APSStatus": {...}
+ * }
+ *
  */
 @Data
 @Builder
@@ -23,6 +33,12 @@ import java.util.List;
 @AllArgsConstructor
 public class ShipDetailApiResponse {
 
+    /**
+     * 선박 개수
+     * API에서 "shipCount"로 반환
+     */
+    @JsonProperty("shipCount")
+    private Integer shipCount;
     /**
      * 선박 상세 정보 리스트
      * API에서 "Ships" (대문자 S)로 반환
@@ -30,10 +46,7 @@ public class ShipDetailApiResponse {
     @JsonProperty("Ships")
     private List<ShipDetailDto> ships;
 
-    /**
-     * 선박 개수
-     * API에서 "shipCount"로 반환
-     */
-    @JsonProperty("shipCount")
-    private Integer shipCount;
+    @JsonProperty("ShipResult")
+    private List<ShipResultDto> shipResult;
+
 }

src/main/java/com/snp/batch/jobs/shipdetail/batch/dto/ShipDetailComparisonData.java

@@ -0,0 +1,17 @@
+package com.snp.batch.jobs.shipdetail.batch.dto;
+import lombok.Builder;
+import lombok.Getter;
+import java.util.Map;
+
+/**
+ * DB 해시와 API 데이터를 결합하여 Processor로 전달하는 컨테이너 DTO
+ */
+@Getter
+@Builder
+public class ShipDetailComparisonData {
+    private final String imoNumber;
+    private final String previousMasterHash;      // DB에서 조회한 해시값 (비교대상 1)
+    private final Map<String, Object> currentMasterMap; // API JSON을 Map으로 변환/정렬/필터링한 데이터
+    private final String currentMasterHash;       // currentMasterMap으로 생성한 해시값 (비교대상 2)
+    private final ShipDetailDto structuredDto;     // 비즈니스 로직 처리용 DTO
+}

src/main/java/com/snp/batch/jobs/shipdetail/batch/dto/ShipDetailDto.java

@@ -1,15 +1,17 @@
 package com.snp.batch.jobs.shipdetail.batch.dto;
 
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
+import java.util.List;
+
 /**
  * 선박 상세 정보 DTO
  * Maritime API GetShipsByIHSLRorIMONumbers 응답 데이터
- *
  * API 응답 필드명과 매핑:
  * - IHSLRorIMOShipNo → imoNumber
  * - ShipName → shipName
@@ -23,6 +25,7 @@ import lombok.NoArgsConstructor;
 @Builder
 @NoArgsConstructor
 @AllArgsConstructor
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class ShipDetailDto {
 
     /**
@@ -30,7 +33,13 @@ public class ShipDetailDto {
      * API: IHSLRorIMOShipNo
      */
     @JsonProperty("IHSLRorIMOShipNo")
-    private String imoNumber;
+    private String ihslrorimoshipno;
+    /**
+     * MMSI 번호
+     * API: MaritimeMobileServiceIdentityMMSINumber
+     */
+    @JsonProperty("MaritimeMobileServiceIdentityMMSINumber")
+    private String maritimemobileserviceidentitymmsinumber;
 
     /**
      * 선박명
@@ -40,32 +49,87 @@ public class ShipDetailDto {
     private String shipName;
 
     /**
-     * 선박 타입
-     * API: ShiptypeLevel5
+     * 호출신호
+     * API: CallSign
      */
-    @JsonProperty("ShiptypeLevel5")
-    private String shipType;
+    @JsonProperty("CallSign")
+    private String callsign;
 
     /**
      * 깃발 국가 (Flag)
      * API: FlagName
      */
     @JsonProperty("FlagName")
-    private String flag;
+    private String flagname;
 
     /**
-     * 깃발 국가 코드
-     * API: FlagCode
+     * 등록항
+     * API: PortOfRegistry
      */
-    @JsonProperty("FlagCode")
-    private String flagCode;
+    @JsonProperty("PortOfRegistry")
+    private String portofregistry;
 
     /**
-     * 깃발 유효 날짜
-     * API: FlagEffectiveDate
+     * 선급
+     * API: ClassificationSociety
      */
-    @JsonProperty("FlagEffectiveDate")
-    private String flagEffectiveDate;
+    @JsonProperty("ClassificationSociety")
+    private String ClassificationSociety;
+
+    /**
+     * 선종(Lv5)
+     * API: ShiptypeLevel5
+     */
+    @JsonProperty("ShiptypeLevel5")
+    private String shiptypelevel5;
+    /**
+     * 선종(Lv2)
+     * API: ShiptypeLevel2
+     */
+    @JsonProperty("ShiptypeLevel2")
+    private String shiptypelevel2;
+    /**
+     * 세부선종
+     * API: ShiptypeLevel5SubType
+     */
+    @JsonProperty("ShiptypeLevel5SubType")
+    private String shiptypelevel5subtype;
+    /**
+     * 건조연도
+     * API: YearOfBuild
+     */
+    @JsonProperty("YearOfBuild")
+    private String yearofbuild;
+    /**
+     * 조선소
+     * API: Shipbuilder
+     */
+    @JsonProperty("Shipbuilder")
+    private String shipbuilder;
+    /**
+     * 전장(LOA)[m]
+     * API: LengthOverallLOA
+     */
+    @JsonProperty("LengthOverallLOA")
+    private String lengthoverallloa;
+    /**
+     * 형폭(몰디드)[m]
+     * API: BreadthMoulded
+     */
+    @JsonProperty("BreadthMoulded")
+    private String breadthmoulded;
+    /**
+     * 깊이[m]
+     * API: Depth
+     */
+    @JsonProperty("Depth")
+    private String depth;
+    /**
+     * 흘수[m]
+     * API: Draught
+     */
+    @JsonProperty("Draught")
+    private String draught;
 
     /**
      * 총톤수 (Gross Tonnage)
@@ -82,23 +146,128 @@ public class ShipDetailDto {
     private String deadweight;
 
     /**
-     * 선박 상태
-     * API: ShipStatus
+     * 컨테이너(TEU)
+     * API: TEU
      */
+    @JsonProperty("TEU")
+    private String teu;
+
+    /**
+     * 항속(kt)
+     * API: SpeedService
+     */
+    @JsonProperty("SpeedService")
+    private String speedservice;
+
+    /**
+     * 주기관 형식
+     * API: MainEngineType
+     */
+    @JsonProperty("MainEngineType")
+    private String mainenginetype;
+
+
     @JsonProperty("ShipStatus")
-    private String status;
+    private String shipStatus;
+
+    @JsonProperty("Operator")
+    private String operator;
+
+    @JsonProperty("FlagCode")
+    private String flagCode;
 
     /**
-     * Core Ship Indicator
-     * API: CoreShipInd
+     * 소유주 이력 List
+     * API: OwnerHistory
      */
-    @JsonProperty("CoreShipInd")
-    private String coreShipInd;
+    @JsonProperty("OwnerHistory")
+    private List<OwnerHistoryDto> ownerHistory;
+
 
     /**
-     * 이전 선박명
-     * API: ExName
+     *  TODO : Core20 Dto 작성
+     * 	shipresultindex 		int8 NOT NULL, -- 결과인덱스
+     * 	batch_flag 				varchar(1) DEFAULT 'N'::character varying NULL -- 업데이트 이력 확인 (N:대기,P:진행,S:완료)
+     * 	vesselid 				varchar(7) NOT NULL, -- 선박ID
+     * 	ihslrorimoshipno 		varchar(7) NOT NULL, -- IMO번호
+     * 	maritimemobileserviceidentitymmsinumber varchar(9) NULL, -- MMSI
+     * 	shipname 				varchar(100) NULL, -- 선명
+     * 	callsign 				varchar(5) NULL, -- 호출부호
+     * 	flagname 				varchar(100) NULL, -- 국가
+     * 	portofregistry 			varchar(50) NULL, -- 등록항
+     * 	classificationsociety 	varchar(50) NULL, -- 선급
+     * 	shiptypelevel5 			varchar(15) NULL, -- 선종(Lv5)
+     * 	shiptypelevel5subtype 	varchar(15) NULL, -- 세부선종
+     * 	yearofbuild 			varchar(4) NULL, -- 건조연도
+     * 	shipbuilder 			varchar(100) NULL, -- 조선소
+     * 	lengthoverallloa 		numeric(3, 3) NULL, -- 전장(LOA)[m]
+     * 	breadthmoulded 			numeric(3, 3) NULL, -- 형폭(몰디드)[m]
+     * 	"depth" 				numeric(3, 3) NULL, -- 깊이[m]
+     * 	draught 				numeric(3, 3) NULL, -- 흘수[m]
+     * 	grosstonnage 			varchar(4) NULL, -- 총톤수(GT)
+     * 	deadweight 				varchar(5) NULL, -- 재화중량톤수(DWT)
+     * 	teu 					varchar(1) NULL, -- 컨테이너(TEU)
+     * 	speedservice 			numeric(2, 2) NULL, -- 항속(kt)
+     * 	mainenginetype 			varchar(2) NULL, -- 주기관 형식
      */
-    @JsonProperty("ExName")
-    private String exName;
+    // TODO : List/Array 데이터 추가
+    /**
+     * 선박 추가 정보 List
+     * API: AdditionalInformation
+     */
+//    @JsonProperty("AdditionalInformation")
+//    private List<ShipAddInfoDto> additionalInformation;
+
+
+    /**
+     * auxengine
+     * auxgenerator
+     * ballastwatermanagement
+     * bareboatcharterhistory
+     * builderaddress
+     * callsignandmmsihistory
+     * capacities
+     * cargopump
+     * classcurrent
+     * classhistory
+     * companycompliancedetails
+     * companydetailscomplexwithcodesa
+     * companyfleetcounts
+     * companyorderbookcounts
+     * companyvesselrelationships
+     * crewlist
+     * darkactivityconfirmed
+     * enginebuilder
+     * flaghistory
+     * grosstonnagehistory
+     * groupbeneficialownerhistory
+     * iceclass
+     * liftinggear
+     * mainengine
+     * namehistory
+     * operatorhistory
+     * ownerhistory
+     * pandihistory
+     * propellers
+     * safetymanagementcertificatehist
+     * sales
+     * scrubberdetails
+     * shipbuilderandsubcontractor
+     * shipbuilderdetail
+     * shipbuilderhistory
+     * shipcertificatesall
+     * shipmanagerhistory
+     * shiptypehistory
+     * sistershiplinks
+     * specialfeature
+     * statushistory
+     * stowagecommodity
+     * surveydates
+     * surveydateshistoryunique
+     * tankcoatings
+     * technicalmanagerhistory
+     * thrusters
+     * */
+
+
 }

src/main/java/com/snp/batch/jobs/shipdetail/batch/dto/ShipDetailUpdate.java

@@ -0,0 +1,28 @@
+package com.snp.batch.jobs.shipdetail.batch.dto;
+
+import com.snp.batch.jobs.shipdetail.batch.entity.OwnerHistoryEntity;
+import com.snp.batch.jobs.shipdetail.batch.entity.ShipDetailEntity;
+import com.snp.batch.jobs.shipdetail.batch.entity.ShipHashEntity;
+import lombok.Builder;
+import lombok.Getter;
+
+import java.util.*;
+
+/**
+ * 변경이 감지되어 Writer로 전달될 최종 증분 업데이트 데이터
+ */
+@Getter
+@Builder
+public class ShipDetailUpdate {
+    private final String imoNumber;
+    private final Map<String, Object> currentMasterMap; // DB JSONB 컬럼 업데이트용 데이터
+    private final String newMasterHash;
+
+    // Hash Update Entity
+    private final ShipHashEntity shipHashEntity;
+
+    // 이 외에 OwnerHistory Entity, Core Entity 등 증분 데이터를 추가합니다.
+    private final ShipDetailEntity shipDetailEntity;
+    private final List<OwnerHistoryEntity> ownerHistoryEntityList;
+
+}

src/main/java/com/snp/batch/jobs/shipdetail/batch/dto/ShipResultDto.java

@@ -0,0 +1,26 @@
+package com.snp.batch.jobs.shipdetail.batch.dto;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.JsonNode;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.*;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ShipResultDto {
+
+    @JsonProperty("shipCount")
+    private Integer shipCount;
+
+    @JsonProperty("APSShipDetail")
+    private JsonNode shipDetailNode;
+    // Getter and Setter
+    public JsonNode getShipDetailNode() { return shipDetailNode; }
+
+}

src/main/java/com/snp/batch/jobs/shipdetail/batch/entity/OwnerHistoryEntity.java

@@ -0,0 +1,23 @@
+package com.snp.batch.jobs.shipdetail.batch.entity;
+
+import com.snp.batch.common.batch.entity.BaseEntity;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+@Data
+@SuperBuilder
+@NoArgsConstructor
+@AllArgsConstructor
+@EqualsAndHashCode(callSuper = true)
+public class OwnerHistoryEntity extends BaseEntity {
+
+    private String CompanyStatus;
+    private String EffectiveDate;
+    private String LRNO;
+    private String Owner;
+    private String OwnerCode;
+    private String Sequence;
+}

src/main/java/com/snp/batch/jobs/shipdetail/batch/entity/ShipDetailEntity.java

@@ -24,75 +24,185 @@ import lombok.experimental.SuperBuilder;
 @EqualsAndHashCode(callSuper = true)
 public class ShipDetailEntity extends BaseEntity {
 
+    /**
+     *  TODO : Core20 Dto 작성
+     * 	shipresultindex 		int8 NOT NULL, -- 결과인덱스
+     * 	ihslrorimoshipno 		varchar(7) NOT NULL, -- IMO번호
+     * 	vesselid 				varchar(7) NOT NULL, -- 선박ID
+     * 	maritimemobileserviceidentitymmsinumber varchar(9) NULL, -- MMSI
+     * 	shipname 				varchar(100) NULL, -- 선명
+     * 	callsign 				varchar(5) NULL, -- 호출부호
+     * 	flagname 				varchar(100) NULL, -- 국가
+     * 	portofregistry 			varchar(50) NULL, -- 등록항
+     * 	classificationsociety 	varchar(50) NULL, -- 선급
+     * 	shiptypelevel5 			varchar(15) NULL, -- 선종(Lv5)
+     * 	shiptypelevel5subtype 	varchar(15) NULL, -- 세부선종
+     * 	yearofbuild 			varchar(4) NULL, -- 건조연도
+     * 	shipbuilder 			varchar(100) NULL, -- 조선소
+     * 	lengthoverallloa 		numeric(3, 3) NULL, -- 전장(LOA)[m]
+     * 	breadthmoulded 			numeric(3, 3) NULL, -- 형폭(몰디드)[m]
+     * 	"depth" 				numeric(3, 3) NULL, -- 깊이[m]
+     * 	draught 				numeric(3, 3) NULL, -- 흘수[m]
+     * 	grosstonnage 			varchar(4) NULL, -- 총톤수(GT)
+     * 	deadweight 				varchar(5) NULL, -- 재화중량톤수(DWT)
+     * 	teu 					varchar(1) NULL, -- 컨테이너(TEU)
+     * 	speedservice 			numeric(2, 2) NULL, -- 항속(kt)
+     * 	mainenginetype 			varchar(2) NULL, -- 주기관 형식
+     * 	batch_flag 				varchar(1) DEFAULT 'N'::character varying NULL -- 업데이트 이력 확인 (N:대기,P:진행,S:완료)
+     */
+
     /**
      * 기본 키 (자동 생성)
      * 컬럼: id (BIGSERIAL)
      */
     private Long id;
 
+    /**
+     * 결과인덱스
+     * 컬럼: shipresultindex (int8, NOT NULL)
+     */
+    private long shipResultIndex;
+
     /**
      * IMO 번호 (비즈니스 키)
      * 컬럼: imo_number (VARCHAR(20), UNIQUE, NOT NULL)
      */
-    private String imoNumber;
+    private String ihslrorimoshipno;
 
     /**
-     * 선박명
-     * 컬럼: ship_name (VARCHAR(200))
+     * 선박ID
+     * 컬럼: vesselid (VARCHAR(7), NOT NULL)
+     */
+    private String vesselId;
+
+    /**
+     * MMSI
+     * 컬럼: maritimemobileserviceidentitymmsinumber (VARCHAR(9))
+     */
+    // Core20 테이블 컬럼 추가
+    private String maritimeMobileServiceIdentityMmsiNumber;
+
+    /**
+     * 선명
+     * 컬럼: shipname (VARCHAR(100))
      */
     private String shipName;
 
     /**
-     * 선박 타입
-     * 컬럼: ship_type (VARCHAR(100))
+     * 호출부호
+     * 컬럼: callsign (VARCHAR(5))
      */
-    private String shipType;
+    private String callSign;
 
     /**
-     * 깃발 국가명
-     * 컬럼: flag_name (VARCHAR(100))
+     * 국가
+     * 컬럼: flagname (VARCHAR(100))
      */
-    private String flag;
+    private String flagName;
 
     /**
-     * 깃발 국가 코드
-     * 컬럼: flag_code (VARCHAR(10))
+     * 등록항
+     * 컬럼: portofregistry (VARCHAR(50))
      */
-    private String flagCode;
+    private String portOfRegistry;
 
     /**
-     * 깃발 유효 날짜
-     * 컬럼: flag_effective_date (VARCHAR(20))
+     * 선급
+     * 컬럼: classificationsociety (VARCHAR(50))
      */
-    private String flagEffectiveDate;
+    private String classificationSociety;
 
     /**
-     * 총톤수 (Gross Tonnage)
-     * 컬럼: gross_tonnage (VARCHAR(20))
+     * 선종(Lv5)
+     * 컬럼: shiptypelevel5 (VARCHAR(15))
+     */
+    private String shipTypeLevel5;
+
+    /**
+     * 선종(Lv2)
+     * 컬럼: shiptypelevel2 (VARCHAR(100))
+     */
+    private String shipTypeLevel2;
+
+    /**
+     * 세부선종
+     * 컬럼: shiptypelevel5subtype (VARCHAR(15))
+     */
+    private String shipTypeLevel5SubType;
+
+    /**
+     * 건조연도
+     * 컬럼: yearofbuild (VARCHAR(4))
+     */
+    private String yearOfBuild;
+
+    /**
+     * 조선소
+     * 컬럼: shipbuilder (VARCHAR(100))
+     */
+    private String shipBuilder;
+
+    /**
+     * 전장(LOA)[m]
+     * 컬럼: lengthoverallloa (NUMERIC(3, 3))
+     */
+    private Double lengthOverallLoa;
+
+    /**
+     * 형폭(몰디드)[m]
+     * 컬럼: breadthmoulded (NUMERIC(3, 3))
+     */
+    private Double breadthMoulded;
+
+    /**
+     * 깊이[m]
+     * 컬럼: depth (NUMERIC(3, 3))
+     */
+    private Double depth;
+
+    /**
+     * 흘수[m]
+     * 컬럼: draught (NUMERIC(3, 3))
+     */
+    private Double draught;
+
+    /**
+     * 총톤수(GT)
+     * 컬럼: grosstonnage (VARCHAR(4))
      */
     private String grossTonnage;
 
     /**
-     * 재화중량톤수 (Deadweight)
-     * 컬럼: deadweight (VARCHAR(20))
+     * 재화중량톤수(DWT)
+     * 컬럼: deadweight (VARCHAR(5))
      */
-    private String deadweight;
+    private String deadWeight;
 
     /**
-     * 선박 상태
-     * 컬럼: ship_status (VARCHAR(100))
+     * 컨테이너(TEU)
+     * 컬럼: teu (VARCHAR(1))
      */
-    private String status;
+    private String teu;
 
     /**
-     * Core Ship Indicator
-     * 컬럼: core_ship_ind (VARCHAR(10))
+     * 항속(kt)
+     * 컬럼: speedservice (NUMERIC(2, 2))
      */
-    private String coreShipInd;
+    private Double speedService;
 
     /**
-     * 이전 선박명
-     * 컬럼: ex_name (VARCHAR(200))
+     * 주기관 형식
+     * 컬럼: mainenginetype (VARCHAR(2))
      */
-    private String exName;
+    private String mainEngineType;
+
+    /**
+     * 업데이트 이력 확인 (N:대기,P:진행,S:완료)
+     * 컬럼: batch_flag (VARCHAR(1))
+    */
+    private String batchFlag;
+    private String shipStatus;
+    private String operator;
+    private String flagCode;
+
 }

src/main/java/com/snp/batch/jobs/shipdetail/batch/entity/ShipHashEntity.java

@@ -0,0 +1,20 @@
+package com.snp.batch.jobs.shipdetail.batch.entity;
+
+import com.snp.batch.common.batch.entity.BaseEntity;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+@Data
+@SuperBuilder
+@NoArgsConstructor
+@AllArgsConstructor
+@EqualsAndHashCode(callSuper = true)
+public class ShipHashEntity extends BaseEntity {
+
+    private String imoNumber;
+    private String shipDetailHash;
+
+}

src/main/java/com/snp/batch/jobs/shipdetail/batch/processor/ShipDetailDataProcessor.java

@@ -1,41 +1,168 @@
 package com.snp.batch.jobs.shipdetail.batch.processor;
 
 import com.snp.batch.common.batch.processor.BaseProcessor;
+import com.snp.batch.jobs.shipdetail.batch.dto.OwnerHistoryDto;
+import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailComparisonData;
 import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailDto;
+import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailUpdate;
+import com.snp.batch.jobs.shipdetail.batch.entity.OwnerHistoryEntity;
 import com.snp.batch.jobs.shipdetail.batch.entity.ShipDetailEntity;
+import com.snp.batch.jobs.shipdetail.batch.entity.ShipHashEntity;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
+import java.util.*;
+
 /**
  * 선박 상세 정보 Processor
  * ShipDetailDto → ShipDetailEntity 변환
  */
+
+/**
+ * 선박 상세 정보 Processor (해시 비교 및 증분 데이터 추출)
+ * I: ShipDetailComparisonData (DB 해시 + API Map Data)
+ * O: ShipDetailUpdate (변경분)
+ */
 @Slf4j
 @Component
-public class ShipDetailDataProcessor extends BaseProcessor<ShipDetailDto, ShipDetailEntity> {
+public class ShipDetailDataProcessor extends BaseProcessor<ShipDetailComparisonData, ShipDetailUpdate> {
 
     @Override
-    protected ShipDetailEntity processItem(ShipDetailDto dto) throws Exception {
-        log.debug("선박 상세 정보 처리 시작: imoNumber={}, shipName={}",
-                dto.getImoNumber(), dto.getShipName());
+    protected ShipDetailUpdate processItem(ShipDetailComparisonData comparisonData) throws Exception {
+        String imo = comparisonData.getImoNumber();
+        String previousHash = comparisonData.getPreviousMasterHash();
+        String currentHash = comparisonData.getCurrentMasterHash();
 
-        // DTO → Entity 변환 (API에서 제공하는 필드만)
-        ShipDetailEntity entity = ShipDetailEntity.builder()
-                .imoNumber(dto.getImoNumber())
-                .shipName(dto.getShipName())
-                .shipType(dto.getShipType())
-                .flag(dto.getFlag())
-                .flagCode(dto.getFlagCode())
-                .flagEffectiveDate(dto.getFlagEffectiveDate())
-                .grossTonnage(dto.getGrossTonnage())
-                .deadweight(dto.getDeadweight())
-                .status(dto.getStatus())
-                .coreShipInd(dto.getCoreShipInd())
-                .exName(dto.getExName())
-                .build();
+        log.debug("선박 해시 비교 시작: imoNumber={}, DB Hash={}, API Hash={}", imo, previousHash, currentHash);
 
-        log.debug("선박 상세 정보 처리 완료: imoNumber={}", dto.getImoNumber());
+        // 1. 해시 비교 (변경 감지)
+        if (isChanged(previousHash, currentHash)) {
+            log.info("선박 데이터 변경 감지: imoNumber={}", imo);
 
-        return entity;
+            // 2. 증분 데이터 추출 (Map 기반으로 업데이트 데이터 구성)
+            // Core20
+            ShipDetailEntity shipDetailEntity = makeShipDetailEntity(comparisonData.getStructuredDto());
+            // OwnerHistory
+            List<OwnerHistoryEntity> ownerHistoryEntityList =  makeOwnerHistoryEntityList(comparisonData.getStructuredDto().getOwnerHistory());
+
+
+
+            log.info("선박 데이터: shipDetailEntity={}", shipDetailEntity.toString());
+            log.info("선박 데이터: ownerHistoryEntityList={}", ownerHistoryEntityList.toString());
+
+            // 3. 최종 업데이트 DTO 생성 (Writer에 전달)
+            return ShipDetailUpdate.builder()
+                    .imoNumber(imo)
+                    .newMasterHash(currentHash) // 새로운 해시
+                    .currentMasterMap(comparisonData.getCurrentMasterMap()) // DB JSONB 컬럼 업데이트용 Map
+                    .shipDetailEntity(shipDetailEntity)
+                    .ownerHistoryEntityList(ownerHistoryEntityList)
+                    .shipHashEntity(makeShipHashEntity(imo, currentHash))
+                    .build();
+        }
+
+        // 변경 사항이 없으면 null 반환 (Writer로 전달되지 않음)
+        return null;
     }
+
+    private ShipHashEntity makeShipHashEntity(String imo, String hash){
+        return ShipHashEntity.builder()
+                .imoNumber(imo)
+                .shipDetailHash(hash)
+                .build();
+    }
+
+    private ShipDetailEntity makeShipDetailEntity(ShipDetailDto dto){
+        return ShipDetailEntity.builder()
+                .ihslrorimoshipno(safeGetString(dto.getIhslrorimoshipno()))
+                .shipName(safeGetString(dto.getShipName()))
+                .vesselId(safeGetString(dto.getIhslrorimoshipno()))
+                .maritimeMobileServiceIdentityMmsiNumber(safeGetString(dto.getMaritimemobileserviceidentitymmsinumber()))
+                .shipName(safeGetString(dto.getShipName()))
+                .callSign(safeGetString(dto.getCallsign()))
+                .flagName(safeGetString(dto.getFlagname()))
+                .portOfRegistry(safeGetString(dto.getPortofregistry()))
+                .classificationSociety(safeGetString(dto.getClassificationSociety()))
+                .shipTypeLevel5(safeGetString(dto.getShiptypelevel5()))
+                .shipTypeLevel2(safeGetString(dto.getShiptypelevel2()))
+                .shipTypeLevel5SubType(safeGetString(dto.getShiptypelevel5subtype()))
+                .yearOfBuild(safeGetString(dto.getYearofbuild()))
+                .shipBuilder(safeGetString(dto.getShipbuilder()))
+                .lengthOverallLoa(safeGetDouble(dto.getLengthoverallloa()))
+                .breadthMoulded(safeGetDouble(dto.getBreadthmoulded()))
+                .depth(safeGetDouble(dto.getDepth()))
+                .draught(safeGetDouble(dto.getDraught()))
+                .grossTonnage(safeGetString(dto.getGrossTonnage()))
+                .deadWeight(safeGetString(dto.getDeadweight()))
+                .teu(safeGetString(dto.getTeu()))
+                .speedService(safeGetDouble(dto.getSpeedservice()))
+                .mainEngineType(safeGetString(dto.getMainenginetype()))
+                .shipStatus(safeGetString(dto.getShipStatus()))
+                .operator(safeGetString(dto.getOperator()))
+                .flagCode(safeGetString(dto.getFlagCode()))
+                .build();
+    }
+
+    private List<OwnerHistoryEntity> makeOwnerHistoryEntityList(List<OwnerHistoryDto> dtoList){
+        List<OwnerHistoryEntity> ownerHistoryEntityList = new ArrayList<>();
+
+        // TODO: ownerHistoryEntityList 생성 로직
+        if (dtoList == null || dtoList.isEmpty()) {
+            return ownerHistoryEntityList; // 빈 리스트 또는 null 입력 시 빈 리스트 반환
+        }
+
+        for (OwnerHistoryDto dto : dtoList) {
+            OwnerHistoryEntity entity = OwnerHistoryEntity.builder()
+                    .CompanyStatus(safeGetString(dto.getCompanyStatus()))
+                    .EffectiveDate(safeGetString(dto.getEffectiveDate()))
+                    .LRNO(safeGetString(dto.getLRNO()))
+                    .Owner(safeGetString(dto.getOwner()))
+                    .OwnerCode(safeGetString(dto.getOwnerCode()))
+                    .Sequence(safeGetString(dto.getSequence()))
+                    .build();
+
+            ownerHistoryEntityList.add(entity);
+        }
+
+        return ownerHistoryEntityList;
+    }
+
+
+
+    /**
+     * 해시값을 비교하여 변경 여부를 판단합니다.
+     */
+    private boolean isChanged(String previousHash, String currentHash) {
+        // DB 해시가 null인 경우 (첫 Insert)는 무조건 변경된 것으로 간주
+        if (previousHash == null || previousHash.isEmpty()) {
+            return true;
+        }
+
+        // 해시값이 다르면 변경된 것으로 간주
+        return !Objects.equals(previousHash, currentHash);
+    }
+
+    /**
+     * DTO 필드 값이 null 또는 빈 문자열(empty/blank)인 경우 null을 반환합니다.
+     * Entity의 String 필드가 null을 허용한다면 이 함수를 사용하세요.
+     */
+    private String safeGetString(String value) {
+        if (value == null || value.trim().isEmpty()) {
+            return null;
+        }
+        // 값이 존재하면 트림된 문자열을 반환합니다.
+        return value.trim();
+    }
+
+    private Double safeGetDouble(String value) {
+        if (value == null || value.trim().isEmpty()) {
+            return null;
+        }
+        try {
+            return Double.parseDouble(value);
+        } catch (NumberFormatException e) {
+            return null;
+        }
+    }
+
 }

src/main/java/com/snp/batch/jobs/shipdetail/batch/reader/ShipDetailDataReader.java

@@ -1,8 +1,13 @@
 package com.snp.batch.jobs.shipdetail.batch.reader;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.snp.batch.common.batch.reader.BaseApiReader;
+import com.snp.batch.common.util.JsonChangeDetector;
 import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailApiResponse;
+import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailComparisonData;
 import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailDto;
+import com.snp.batch.jobs.shipdetail.batch.dto.ShipResultDto;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.web.reactive.function.client.WebClient;
@@ -30,18 +35,22 @@ import java.util.*;
  * - 신규: 100건씩 로드 → Process → Write (Chunk 1,718회)
  */
 @Slf4j
-public class ShipDetailDataReader extends BaseApiReader<ShipDetailDto> {
+public class ShipDetailDataReader extends BaseApiReader<ShipDetailComparisonData> {
 
     private final JdbcTemplate jdbcTemplate;
+    private final ObjectMapper objectMapper;
 
     // 배치 처리 상태
     private List<String> allImoNumbers;
+    // DB 해시값을 저장할 맵
+    private Map<String, String> dbMasterHashes;
     private int currentBatchIndex = 0;
-    private final int batchSize = 100;
+    private final int batchSize = 50;
 
-    public ShipDetailDataReader(WebClient webClient, JdbcTemplate jdbcTemplate) {
+    public ShipDetailDataReader(WebClient webClient, JdbcTemplate jdbcTemplate, ObjectMapper objectMapper) {
         super(webClient);
         this.jdbcTemplate = jdbcTemplate;
+        this.objectMapper = objectMapper;
         enableChunkMode();  // ✨ Chunk 모드 활성화
     }
 
@@ -52,7 +61,7 @@ public class ShipDetailDataReader extends BaseApiReader<ShipDetailDto> {
 
     @Override
     protected String getApiPath() {
-        return "/MaritimeWCF/APSShipService.svc/RESTFul/GetShipsByIHSLRorIMONumbers";
+        return "/MaritimeWCF/APSShipService.svc/RESTFul/GetShipsByIHSLRorIMONumbersAll";
     }
 
     @Override
@@ -60,22 +69,42 @@ public class ShipDetailDataReader extends BaseApiReader<ShipDetailDto> {
         return "https://shipsapi.maritime.spglobal.com";
     }
 
+    private static final String GET_ALL_IMO_QUERY =
+            "SELECT imo_number FROM ship_data ORDER BY id";
+
+    private static final String FETCH_ALL_HASHES_QUERY =
+            "SELECT imo_number, ship_detail_hash FROM ship_detail_hash_json ORDER BY imo_number";
+
     /**
      * 최초 1회만 실행: ship_data 테이블에서 IMO 번호 전체 조회
      */
     @Override
     protected void beforeFetch() {
+        // 전처리 과정
+        // Step 1. IMO 전체 번호 조회
         log.info("[{}] ship_data 테이블에서 IMO 번호 조회 시작...", getReaderName());
 
-        String sql = "SELECT imo_number FROM ship_data ORDER BY id";
-        allImoNumbers = jdbcTemplate.queryForList(sql, String.class);
-
+        allImoNumbers = jdbcTemplate.queryForList(GET_ALL_IMO_QUERY, String.class);
         int totalBatches = (int) Math.ceil((double) allImoNumbers.size() / batchSize);
 
         log.info("[{}] 총 {} 개의 IMO 번호 조회 완료", getReaderName(), allImoNumbers.size());
         log.info("[{}] {}개씩 배치로 분할하여 API 호출 예정", getReaderName(), batchSize);
         log.info("[{}] 예상 배치 수: {} 개", getReaderName(), totalBatches);
 
+        // Step 2. 전 배치 결과 imo_number, ship_detail_json, ship_detail_hash 데이터 전체 조회
+        log.info("[{}] DB Master Hash 전체 조회 시작...", getReaderName());
+
+        // 1-1. DB에서 모든 IMO와 Hash 조회
+        dbMasterHashes = jdbcTemplate.query(FETCH_ALL_HASHES_QUERY, rs -> {
+            Map<String, String> map = new HashMap<>();
+            while (rs.next()) {
+                map.put(rs.getString("imo_number"), rs.getString("ship_detail_hash"));
+            }
+            return map;
+        });
+
+        log.info("[{}] DB Master Hash 조회 완료. 총 {}건.", getReaderName(), dbMasterHashes.size());
+
         // API 통계 초기화
         updateApiCallStats(totalBatches, 0);
     }
@@ -88,7 +117,8 @@ public class ShipDetailDataReader extends BaseApiReader<ShipDetailDto> {
      * @return 다음 배치 100건 (더 이상 없으면 null)
      */
     @Override
-    protected List<ShipDetailDto> fetchNextBatch() throws Exception {
+    protected List<ShipDetailComparisonData> fetchNextBatch() throws Exception {
+
         // 모든 배치 처리 완료 확인
         if (allImoNumbers == null || currentBatchIndex >= allImoNumbers.size()) {
             return null; // Job 종료
@@ -117,11 +147,50 @@ public class ShipDetailDataReader extends BaseApiReader<ShipDetailDto> {
             // 다음 배치로 인덱스 이동
             currentBatchIndex = endIndex;
 
+            List<ShipDetailComparisonData> comparisonList = new ArrayList<>();
+
             // 응답 처리
-            if (response != null && response.getShips() != null) {
-                List<ShipDetailDto> ships = response.getShips();
+            if (response != null && response.getShipResult() != null) {
+                // ✨ ShipDetailComparisonData 생성
+                for (ShipResultDto shipResult : response.getShipResult()) {
+                    JsonNode rawShipDetailNode = shipResult.getShipDetailNode();
+
+                    if (rawShipDetailNode == null) continue;
+
+                    // 💡 1) DTO로 매핑하여 IMO와 구조화된 DTO 확보
+                    ShipDetailDto structuredDto = null;
+                    try {
+                        structuredDto = objectMapper.treeToValue(rawShipDetailNode, ShipDetailDto.class);
+                    } catch (Exception e) {
+                        log.error("ShipDetailDto 매핑 실패: {}", e.getMessage());
+                        continue;
+                    }
+
+                    String imo = structuredDto.getIhslrorimoshipno();
+                    if (imo == null || imo.isEmpty()) continue;
+
+                    // 💡 2) 원본 JSON 문자열 생성
+                    String originalJsonString = rawShipDetailNode.toString();
+
+                    // 💡 3) API response json을 Map 형태로 변환, 정렬 및 해시 생성
+                    Map<String, Object> currentMasterMap = JsonChangeDetector.jsonToSortedFilteredMap(originalJsonString);
+                    String currentMasterHash = JsonChangeDetector.getSha256HashFromMap(currentMasterMap);
+
+                    // 💡 4) DB Master Hash 조회 (beforeFetch에서 로드된 맵 사용)
+                    String previousMasterHash = dbMasterHashes.getOrDefault(imo, null);
+
+                    // 💡 5) ShipDetailComparisonData DTO 생성
+                    comparisonList.add(ShipDetailComparisonData.builder()
+                            .imoNumber(imo)
+                            .previousMasterHash(previousMasterHash)
+                            .currentMasterMap(currentMasterMap)
+                            .currentMasterHash(currentMasterHash)
+                            .structuredDto(structuredDto)
+                            .build());
+                }
+
                 log.info("[{}] 배치 {}/{} 완료: {} 건 조회",
-                        getReaderName(), currentBatchNumber, totalBatches, ships.size());
+                        getReaderName(), currentBatchNumber, totalBatches, comparisonList.size());
 
                 // API 호출 통계 업데이트
                 updateApiCallStats(totalBatches, currentBatchNumber);
@@ -131,7 +200,8 @@ public class ShipDetailDataReader extends BaseApiReader<ShipDetailDto> {
                     Thread.sleep(500);
                 }
 
-                return ships;
+                return comparisonList;
+
             } else {
                 log.warn("[{}] 배치 {}/{} 응답 없음",
                         getReaderName(), currentBatchNumber, totalBatches);
@@ -161,7 +231,7 @@ public class ShipDetailDataReader extends BaseApiReader<ShipDetailDto> {
      * @return API 응답
      */
     private ShipDetailApiResponse callApiWithBatch(String imoNumbers) {
-        String url = getApiPath() + "?ihslrOrImoNumbers=" + imoNumbers;
+        String url = getApiPath() + "?IMONumbers=" + imoNumbers;
 
         log.debug("[{}] API 호출: {}", getReaderName(), url);
 
@@ -173,7 +243,7 @@ public class ShipDetailDataReader extends BaseApiReader<ShipDetailDto> {
     }
 
     @Override
-    protected void afterFetch(List<ShipDetailDto> data) {
+    protected void afterFetch(List<ShipDetailComparisonData> data) {
         if (data == null) {
             int totalBatches = (int) Math.ceil((double) allImoNumbers.size() / batchSize);
             log.info("[{}] 전체 {} 개 배치 처리 완료", getReaderName(), totalBatches);
@@ -181,4 +251,5 @@ public class ShipDetailDataReader extends BaseApiReader<ShipDetailDto> {
                     getReaderName(), allImoNumbers.size());
         }
     }
+
 }

src/main/java/com/snp/batch/jobs/shipdetail/batch/repository/ShipDetailRepository.java

@@ -1,42 +1,21 @@
 package com.snp.batch.jobs.shipdetail.batch.repository;
 
+import com.snp.batch.jobs.shipdetail.batch.entity.OwnerHistoryEntity;
 import com.snp.batch.jobs.shipdetail.batch.entity.ShipDetailEntity;
 
 import java.util.List;
-import java.util.Optional;
 
 /**
  * 선박 상세 정보 Repository 인터페이스
  */
 public interface ShipDetailRepository {
 
-    /**
-     * ID로 조회
-     */
-    Optional<ShipDetailEntity> findById(Long id);
-
-    /**
-     * IMO 번호로 조회
-     */
-    Optional<ShipDetailEntity> findByImoNumber(String imoNumber);
-
-    /**
-     * 전체 조회
-     */
-    List<ShipDetailEntity> findAll();
-
-    /**
-     * 저장 (INSERT 또는 UPDATE)
-     */
-    ShipDetailEntity save(ShipDetailEntity entity);
-
-    /**
-     * 여러 건 저장
-     */
     void saveAll(List<ShipDetailEntity> entities);
 
-    /**
-     * 삭제
-     */
-    void delete(Long id);
+    void saveAllCoreData(List<ShipDetailEntity> entities);
+
+    void saveAllOwnerHistoryData(List<OwnerHistoryEntity> entities);
+
+    void delete(String id);
+
 }

src/main/java/com/snp/batch/jobs/shipdetail/batch/repository/ShipDetailRepositoryImpl.java

@@ -1,6 +1,7 @@
 package com.snp.batch.jobs.shipdetail.batch.repository;
 
 import com.snp.batch.common.batch.repository.BaseJdbcRepository;
+import com.snp.batch.jobs.shipdetail.batch.entity.OwnerHistoryEntity;
 import com.snp.batch.jobs.shipdetail.batch.entity.ShipDetailEntity;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.jdbc.core.JdbcTemplate;
@@ -11,8 +12,7 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Timestamp;
-import java.util.List;
-import java.util.Optional;
+import java.util.*;
 
 /**
  * 선박 상세 정보 Repository 구현체
@@ -20,7 +20,7 @@ import java.util.Optional;
  */
 @Slf4j
 @Repository("shipDetailRepository")
-public class ShipDetailRepositoryImpl extends BaseJdbcRepository<ShipDetailEntity, Long>
+public class ShipDetailRepositoryImpl extends BaseJdbcRepository<ShipDetailEntity, String>
         implements ShipDetailRepository {
 
     public ShipDetailRepositoryImpl(JdbcTemplate jdbcTemplate) {
@@ -29,99 +29,165 @@ public class ShipDetailRepositoryImpl extends BaseJdbcRepository<ShipDetailEntit
 
     @Override
     protected String getTableName() {
-        return "ship_detail";
+        return "test_s_p.test_core20";
     }
 
     @Override
     protected String getEntityName() {
-        return "ShipDetail";
+        return "ShipDetailEntity";
     }
 
     @Override
-    protected Long extractId(ShipDetailEntity entity) {
-        return entity.getId();
+    protected String extractId(ShipDetailEntity entity) {
+        return entity.getIhslrorimoshipno();
     }
 
     @Override
     protected String getInsertSql() {
         return """
-            INSERT INTO ship_detail (
-                imo_number, ship_name, ship_type,
-                flag_name, flag_code, flag_effective_date,
-                gross_tonnage, deadweight,
-                ship_status, core_ship_ind, ex_name,
-                created_at, updated_at, created_by, updated_by
-            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-            ON CONFLICT (imo_number)
-            DO UPDATE SET
-                ship_name = EXCLUDED.ship_name,
-                ship_type = EXCLUDED.ship_type,
-                flag_name = EXCLUDED.flag_name,
-                flag_code = EXCLUDED.flag_code,
-                flag_effective_date = EXCLUDED.flag_effective_date,
-                gross_tonnage = EXCLUDED.gross_tonnage,
-                deadweight = EXCLUDED.deadweight,
-                ship_status = EXCLUDED.ship_status,
-                core_ship_ind = EXCLUDED.core_ship_ind,
-                ex_name = EXCLUDED.ex_name,
-                updated_at = EXCLUDED.updated_at,
-                updated_by = EXCLUDED.updated_by
-            """;
+                INSERT INTO test_s_p.test_core20(
+                        shipresultindex, vesselid, ihslrorimoshipno, maritimemobileserviceidentitymmsinumber, shipname,
+                        callsign, flagname, portofregistry, classificationsociety, shiptypelevel5,
+                        shiptypelevel5subtype, yearofbuild, shipbuilder, lengthoverallloa, breadthmoulded,
+                        "depth", draught, grosstonnage, deadweight, teu,
+                        speedservice, mainenginetype, status, operator, flagcode, shiptypelevel2
+                    ) VALUES (nextval('test_s_p.core20_index_seq'::regclass), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    ON CONFLICT (ihslrorimoshipno)
+                    DO UPDATE SET
+                        vesselid = EXCLUDED.vesselid,
+                        maritimemobileserviceidentitymmsinumber = EXCLUDED.maritimemobileserviceidentitymmsinumber,
+                        shipname = EXCLUDED.shipname,
+                        callsign = EXCLUDED.callsign,
+                        flagname = EXCLUDED.flagname,
+                        portofregistry = EXCLUDED.portofregistry,
+                        classificationsociety = EXCLUDED.classificationsociety,
+                        shiptypelevel5 = EXCLUDED.shiptypelevel5,
+                        shiptypelevel5subtype = EXCLUDED.shiptypelevel5subtype,
+                        yearofbuild = EXCLUDED.yearofbuild,
+                        shipbuilder = EXCLUDED.shipbuilder,
+                        lengthoverallloa = EXCLUDED.lengthoverallloa,
+                        breadthmoulded = EXCLUDED.breadthmoulded,
+                        "depth" = EXCLUDED.depth,
+                        draught = EXCLUDED.draught,
+                        grosstonnage = EXCLUDED.grosstonnage,
+                        deadweight = EXCLUDED.deadweight,
+                        teu = EXCLUDED.teu,
+                        speedservice = EXCLUDED.speedservice,
+                        mainenginetype = EXCLUDED.mainenginetype,
+                        status = EXCLUDED.status,
+                        operator = EXCLUDED.operator,
+                        flagcode = EXCLUDED.flagcode,
+                        shiptypelevel2 = EXCLUDED.shiptypelevel2
+                """;
     }
 
     @Override
     protected String getUpdateSql() {
         return """
-            UPDATE ship_detail
-            SET ship_name = ?, ship_type = ?,
-                flag_name = ?, flag_code = ?, flag_effective_date = ?,
-                gross_tonnage = ?, deadweight = ?,
-                ship_status = ?, core_ship_ind = ?, ex_name = ?,
-                updated_at = ?, updated_by = ?
-            WHERE id = ?
-            """;
+                UPDATE test_s_p.test_core20
+                   SET  vesselid = ?,
+                        maritimemobileserviceidentitymmsinumber = ?,
+                        shipname = ?,
+                        callsign = ?,
+                        flagname = ?,
+                        portofregistry = ?,
+                        classificationsociety = ?,
+                        shiptypelevel5 = ?,
+                        shiptypelevel5subtype = ?,
+                        yearofbuild = ?,
+                        shipbuilder = ?,
+                        lengthoverallloa = ?,
+                        breadthmoulded = ?,
+                        "depth" = ?,
+                        draught = ?,
+                        grosstonnage = ?,
+                        deadweight = ?,
+                        teu = ?,
+                        speedservice = ?,
+                        mainenginetype = ?,
+                        batch_flag = 'N'::character varying,
+                        status = ?,
+                        operator = ?,
+                        flagcode = ?,
+                        shiptypelevel2 = ?
+                WHERE ihslrorimoshipno = ?
+                """;
     }
 
     @Override
     protected void setInsertParameters(PreparedStatement ps, ShipDetailEntity entity) throws Exception {
         int idx = 1;
-        ps.setString(idx++, entity.getImoNumber());
+        ps.setString(idx++, entity.getIhslrorimoshipno()); //vesselId
+        ps.setString(idx++, entity.getIhslrorimoshipno());
+        ps.setString(idx++, entity.getMaritimeMobileServiceIdentityMmsiNumber());
         ps.setString(idx++, entity.getShipName());
-        ps.setString(idx++, entity.getShipType());
-        ps.setString(idx++, entity.getFlag());
-        ps.setString(idx++, entity.getFlagCode());
-        ps.setString(idx++, entity.getFlagEffectiveDate());
+        ps.setString(idx++, entity.getCallSign());
+        ps.setString(idx++, entity.getFlagName());
+        ps.setString(idx++, entity.getPortOfRegistry());
+        ps.setString(idx++, entity.getClassificationSociety());
+        ps.setString(idx++, entity.getShipTypeLevel5());
+        ps.setString(idx++, entity.getShipTypeLevel5SubType());
+        ps.setString(idx++, entity.getYearOfBuild());
+        ps.setString(idx++, entity.getShipBuilder());
+        setDoubleOrNull(ps, idx++, entity.getLengthOverallLoa());
+        setDoubleOrNull(ps, idx++, entity.getBreadthMoulded());
+        setDoubleOrNull(ps, idx++, entity.getDepth());
+        setDoubleOrNull(ps, idx++, entity.getDraught());
         ps.setString(idx++, entity.getGrossTonnage());
-        ps.setString(idx++, entity.getDeadweight());
-        ps.setString(idx++, entity.getStatus());
-        ps.setString(idx++, entity.getCoreShipInd());
-        ps.setString(idx++, entity.getExName());
+        ps.setString(idx++, entity.getDeadWeight());
+        ps.setString(idx++, entity.getTeu());
+        setDoubleOrNull(ps, idx++, entity.getSpeedService());
+        ps.setString(idx++, entity.getMainEngineType());
+        ps.setString(idx++, entity.getShipStatus());
+        ps.setString(idx++, entity.getOperator());
+        ps.setString(idx++, entity.getFlagCode());
+        ps.setString(idx++, entity.getShipTypeLevel2());
 
         // 감사 필드
-        ps.setTimestamp(idx++, entity.getCreatedAt() != null ?
-                Timestamp.valueOf(entity.getCreatedAt()) : Timestamp.valueOf(now()));
-        ps.setTimestamp(idx++, entity.getUpdatedAt() != null ?
-                Timestamp.valueOf(entity.getUpdatedAt()) : Timestamp.valueOf(now()));
-        ps.setString(idx++, entity.getCreatedBy() != null ? entity.getCreatedBy() : "SYSTEM");
-        ps.setString(idx++, entity.getUpdatedBy() != null ? entity.getUpdatedBy() : "SYSTEM");
+//        ps.setTimestamp(idx++, entity.getCreatedAt() != null ?
+//                Timestamp.valueOf(entity.getCreatedAt()) : Timestamp.valueOf(now()));
+//        ps.setTimestamp(idx++, entity.getUpdatedAt() != null ?
+//                Timestamp.valueOf(entity.getUpdatedAt()) : Timestamp.valueOf(now()));
+//        ps.setString(idx++, entity.getCreatedBy() != null ? entity.getCreatedBy() : "SYSTEM");
+//        ps.setString(idx++, entity.getUpdatedBy() != null ? entity.getUpdatedBy() : "SYSTEM");
+    }
+    private void setDoubleOrNull(PreparedStatement ps, int index, Double value) throws Exception {
+        if (value != null) {
+            ps.setDouble(index, value);
+        } else {
+            // java.sql.Types.DOUBLE을 사용하여 명시적으로 SQL NULL을 설정
+            ps.setNull(index, java.sql.Types.DOUBLE);
+        }
     }
 
     @Override
     protected void setUpdateParameters(PreparedStatement ps, ShipDetailEntity entity) throws Exception {
         int idx = 1;
+        ps.setString(idx++, entity.getVesselId());
+        ps.setString(idx++, entity.getMaritimeMobileServiceIdentityMmsiNumber());
         ps.setString(idx++, entity.getShipName());
-        ps.setString(idx++, entity.getShipType());
-        ps.setString(idx++, entity.getFlag());
-        ps.setString(idx++, entity.getFlagCode());
-        ps.setString(idx++, entity.getFlagEffectiveDate());
+        ps.setString(idx++, entity.getCallSign());
+        ps.setString(idx++, entity.getFlagName());
+        ps.setString(idx++, entity.getPortOfRegistry());
+        ps.setString(idx++, entity.getClassificationSociety());
+        ps.setString(idx++, entity.getShipTypeLevel5());
+        ps.setString(idx++, entity.getShipTypeLevel5SubType());
+        ps.setString(idx++, entity.getYearOfBuild());
+        ps.setString(idx++, entity.getShipBuilder());
+        ps.setDouble(idx++, entity.getLengthOverallLoa());
+        ps.setDouble(idx++, entity.getBreadthMoulded());
+        ps.setDouble(idx++, entity.getDepth());
+        ps.setDouble(idx++, entity.getDraught());
         ps.setString(idx++, entity.getGrossTonnage());
-        ps.setString(idx++, entity.getDeadweight());
-        ps.setString(idx++, entity.getStatus());
-        ps.setString(idx++, entity.getCoreShipInd());
-        ps.setString(idx++, entity.getExName());
-        ps.setTimestamp(idx++, Timestamp.valueOf(now()));
-        ps.setString(idx++, entity.getUpdatedBy() != null ? entity.getUpdatedBy() : "SYSTEM");
-        ps.setLong(idx++, entity.getId());
+        ps.setString(idx++, entity.getDeadWeight());
+        ps.setString(idx++, entity.getTeu());
+        ps.setDouble(idx++, entity.getSpeedService());
+        ps.setString(idx++, entity.getMainEngineType());
+        ps.setString(idx++, entity.getShipStatus());
+        ps.setString(idx++, entity.getOperator());
+        ps.setString(idx++, entity.getFlagCode());
+        ps.setString(idx++, entity.getShipTypeLevel2());
+        ps.setString(idx++, entity.getIhslrorimoshipno());
     }
 
     @Override
@@ -130,19 +196,83 @@ public class ShipDetailRepositoryImpl extends BaseJdbcRepository<ShipDetailEntit
     }
 
     @Override
-    public Optional<ShipDetailEntity> findByImoNumber(String imoNumber) {
-        String sql = "SELECT * FROM " + getTableName() + " WHERE imo_number = ?";
-        List<ShipDetailEntity> results = jdbcTemplate.query(sql, getRowMapper(), imoNumber);
-        return results.isEmpty() ? Optional.empty() : Optional.of(results.get(0));
+    public void saveAllCoreData(List<ShipDetailEntity> entities) {
+        if (entities == null || entities.isEmpty()) {
+            return;
+        }
+
+        log.info("{} 전체 저장 시작: {} 건", getEntityName(), entities.size());
+
+        // INSERT와 UPDATE 분리
+        List<ShipDetailEntity> toInsert = entities.stream()
+                .filter(e -> extractId(e) == null || !existsByImo(extractId(e)))
+                .toList();
+
+        List<ShipDetailEntity> toUpdate = entities.stream()
+                .filter(e -> extractId(e) != null && existsByImo(extractId(e)))
+                .toList();
+
+        if (!toInsert.isEmpty()) {
+            batchInsert(toInsert);
+        }
+
+        if (!toUpdate.isEmpty()) {
+            batchUpdate(toUpdate);
+        }
+
+        log.info("{} 전체 저장 완료: 삽입={} 건, 수정={} 건", getEntityName(), toInsert.size(), toUpdate.size());
+
     }
 
     @Override
-    public void delete(Long id) {
-        String sql = "DELETE FROM " + getTableName() + " WHERE id = ?";
+    public void saveAllOwnerHistoryData(List<OwnerHistoryEntity> entities) {
+        String ownerHistorySql = ShipDetailSql.getOwnerHistorySql();
+
+        if (entities == null || entities.isEmpty()) {
+            return;
+        }
+
+        log.info("{} 배치 삽입 시작: {} 건", "OwnerHistoryEntity", entities.size());
+
+        jdbcTemplate.batchUpdate(ownerHistorySql, entities, entities.size(),
+                (ps, entity) -> {
+                    try {
+                        setOwnerHistoryInsertParameters(ps, (OwnerHistoryEntity) entity);
+                    } catch (Exception e) {
+                        log.error("배치 삽입 파라미터 설정 실패", e);
+                        throw new RuntimeException(e);
+                    }
+                });
+
+        log.info("{} 배치 삽입 완료: {} 건", getEntityName(), entities.size());
+    }
+
+    public boolean existsByImo(String imo) {
+        String sql = String.format("SELECT COUNT(*) FROM %s WHERE %s = ?", getTableName(), getIdColumnName("ihslrorimoshipno"));
+        Long count = jdbcTemplate.queryForObject(sql, Long.class, imo);
+        return count != null && count > 0;
+    }
+
+
+
+    @Override
+    public void delete(String id) {
+        String sql = "DELETE FROM " + getTableName() + " WHERE ihslrorimoshipno = ?";
         jdbcTemplate.update(sql, id);
         log.debug("[{}] 삭제 완료: id={}", getEntityName(), id);
     }
 
+    private void setOwnerHistoryInsertParameters(PreparedStatement ps, OwnerHistoryEntity entity)throws Exception{
+        int idx = 1;
+        ps.setString(idx++, entity.getCompanyStatus());
+        ps.setString(idx++, entity.getEffectiveDate());
+        ps.setString(idx++, entity.getLRNO());
+        ps.setString(idx++, entity.getOwner());
+        ps.setString(idx++, entity.getOwnerCode());
+        ps.setString(idx++, entity.getSequence());
+        ps.setString(idx++, entity.getLRNO()); //vesselId
+    }
+
     /**
      * ShipDetailEntity RowMapper
      */
@@ -151,21 +281,34 @@ public class ShipDetailRepositoryImpl extends BaseJdbcRepository<ShipDetailEntit
         public ShipDetailEntity mapRow(ResultSet rs, int rowNum) throws SQLException {
             ShipDetailEntity entity = ShipDetailEntity.builder()
                     .id(rs.getLong("id"))
-                    .imoNumber(rs.getString("imo_number"))
-                    .shipName(rs.getString("ship_name"))
-                    .shipType(rs.getString("ship_type"))
-                    .flag(rs.getString("flag_name"))
-                    .flagCode(rs.getString("flag_code"))
-                    .flagEffectiveDate(rs.getString("flag_effective_date"))
-                    .grossTonnage(rs.getString("gross_tonnage"))
-                    .deadweight(rs.getString("deadweight"))
-                    .status(rs.getString("ship_status"))
-                    .coreShipInd(rs.getString("core_ship_ind"))
-                    .exName(rs.getString("ex_name"))
+                    .shipResultIndex(rs.getLong("shipresultindex"))
+                    .ihslrorimoshipno(rs.getString("ihslrorimoshipno"))
+                    .vesselId(rs.getString("vesselid"))
+                    .maritimeMobileServiceIdentityMmsiNumber(rs.getString("maritimemobileserviceidentitymmsinumber"))
+                    .shipName(rs.getString("shipname"))
+                    .callSign(rs.getString("callsign"))
+                    .flagName(rs.getString("flagname"))
+                    .portOfRegistry(rs.getString("portofregistry"))
+                    .classificationSociety(rs.getString("classificationsociety"))
+                    .shipTypeLevel5(rs.getString("shiptypelevel5"))
+                    .shipTypeLevel5SubType(rs.getString("shiptypelevel5subtype"))
+                    .yearOfBuild(rs.getString("yearofbuild"))
+                    .shipBuilder(rs.getString("shipbuilder"))
+                    .lengthOverallLoa(rs.getDouble("lengthoverallloa"))
+                    .breadthMoulded(rs.getDouble("breadthmoulded"))
+                    .depth(rs.getDouble("depth"))
+                    .draught(rs.getDouble("draught"))
+                    .grossTonnage(rs.getString("grosstonnage"))
+                    .deadWeight(rs.getString("deadweight"))
+                    .teu(rs.getString("teu"))
+                    .speedService(rs.getDouble("speedservice"))
+                    .mainEngineType(rs.getString("mainenginetype"))
+                    .batchFlag(rs.getString("batch_flag"))
+                    .shipTypeLevel2(rs.getString("shiptypelevel2"))
                     .build();
 
             // BaseEntity 필드 매핑
-            Timestamp createdAt = rs.getTimestamp("created_at");
+            Timestamp createdAt = rs.getTimestamp("batch_flag");
             if (createdAt != null) {
                 entity.setCreatedAt(createdAt.toLocalDateTime());
             }

src/main/java/com/snp/batch/jobs/shipdetail/batch/repository/ShipDetailSql.java

@@ -0,0 +1,22 @@
+package com.snp.batch.jobs.shipdetail.batch.repository;
+
+public class ShipDetailSql {
+
+    public static String getOwnerHistorySql(){
+        return """
+               INSERT INTO test_s_p.test_ownerhistory(
+                    datasetversion, companystatus, effectivedate, lrno, "owner", 
+                    ownercode, "sequence", shipresultindex, vesselid, rowindex, batch_flag
+                )VALUES('1.0.0', ?, ?, ?, ?, ?, ?, nextval('test_s_p.ownerhistory_index_seq'::regclass), ?, nextval('test_s_p.ownerhistory_row_index_seq'::regclass), 'N')
+                ON CONFLICT (lrno,ownercode, effectivedate)
+                DO UPDATE SET
+                    datasetversion = EXCLUDED.datasetversion, 
+                    companystatus = EXCLUDED.companystatus,
+                    effectivedate = EXCLUDED.effectivedate, 
+                    "owner" = EXCLUDED.owner, 
+                    ownercode = EXCLUDED.ownercode, 
+                    "sequence" = EXCLUDED.sequence,
+                    batch_flag = 'N'
+               """;
+    }
+}

src/main/java/com/snp/batch/jobs/shipdetail/batch/repository/ShipHashRepository.java

@@ -0,0 +1,10 @@
+package com.snp.batch.jobs.shipdetail.batch.repository;
+
+import com.snp.batch.jobs.shipdetail.batch.entity.ShipHashEntity;
+
+import java.util.List;
+
+public interface ShipHashRepository {
+    void saveAllData(List<ShipHashEntity> entities);
+
+}

src/main/java/com/snp/batch/jobs/shipdetail/batch/repository/ShipHashRepositoryImpl.java

@@ -0,0 +1,129 @@
+package com.snp.batch.jobs.shipdetail.batch.repository;
+
+import com.snp.batch.common.batch.repository.BaseJdbcRepository;
+import com.snp.batch.jobs.shipdetail.batch.entity.ShipHashEntity;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.stereotype.Repository;
+
+import java.sql.PreparedStatement;
+import java.sql.Timestamp;
+import java.util.List;
+
+@Slf4j
+@Repository("ShipHashRepository")
+public class ShipHashRepositoryImpl extends BaseJdbcRepository<ShipHashEntity, String> implements ShipHashRepository{
+
+    public ShipHashRepositoryImpl(JdbcTemplate jdbcTemplate) {
+        super(jdbcTemplate);
+    }
+
+    @Override
+    protected String getTableName() {
+        return "public.ship_detail_hash_json";
+    }
+
+    @Override
+    protected RowMapper<ShipHashEntity> getRowMapper() {
+        return null;
+    }
+
+    @Override
+    protected String extractId(ShipHashEntity entity) {
+        return entity.getImoNumber();
+    }
+
+    @Override
+    protected String getInsertSql() {
+        return """
+                INSERT INTO public.ship_detail_hash_json(
+                    id, imo_number, ship_detail_hash, created_at, created_by, updated_at, updated_by
+                )VALUES(
+                    nextval('ship_imo_data_id_seq1'::regclass), ?, ?, ?, ?, ?, ?
+                )
+                ON CONFLICT (imo_number)
+                DO UPDATE SET
+                   ship_detail_hash = EXCLUDED.ship_detail_hash,
+                   updated_at = ?,
+                   updated_by = ?
+            """;
+    }
+
+    @Override
+    protected String getUpdateSql() {
+        return  """
+                UPDATE public.ship_detail_hash_json
+                   SET ship_detail_hash = ?,
+                       updated_at = ?,
+                       updated_by = ?
+                 WHERE imo_number = ?
+            """;
+    }
+
+    @Override
+    protected void setInsertParameters(PreparedStatement ps, ShipHashEntity entity) throws Exception {
+        int idx = 1;
+        ps.setString(idx++, entity.getImoNumber());
+        ps.setString(idx++, entity.getShipDetailHash());
+        // 감사 필드
+        ps.setTimestamp(idx++, entity.getCreatedAt() != null ?
+                Timestamp.valueOf(entity.getCreatedAt()) : Timestamp.valueOf(now()));
+        ps.setString(idx++, entity.getCreatedBy() != null ? entity.getCreatedBy() : "SYSTEM");
+        ps.setTimestamp(idx++, entity.getUpdatedAt() != null ?
+                Timestamp.valueOf(entity.getUpdatedAt()) : Timestamp.valueOf(now()));
+        ps.setString(idx++, entity.getUpdatedBy() != null ? entity.getUpdatedBy() : "SYSTEM");ps.setTimestamp(idx++, entity.getUpdatedAt() != null ?
+                Timestamp.valueOf(entity.getUpdatedAt()) : Timestamp.valueOf(now()));
+        ps.setString(idx++, entity.getUpdatedBy() != null ? entity.getUpdatedBy() : "SYSTEM");
+    }
+
+    @Override
+    protected void setUpdateParameters(PreparedStatement ps, ShipHashEntity entity) throws Exception {
+        int idx = 1;
+        ps.setString(idx++, entity.getShipDetailHash());
+        // 감사 필드
+        ps.setTimestamp(idx++, entity.getUpdatedAt() != null ?
+                Timestamp.valueOf(entity.getUpdatedAt()) : Timestamp.valueOf(now()));
+        ps.setString(idx++, entity.getUpdatedBy() != null ? entity.getUpdatedBy() : "SYSTEM");
+        ps.setString(idx++, entity.getImoNumber());
+    }
+
+    @Override
+    protected String getEntityName() {
+        return "ShipHashEntity";
+    }
+
+    @Override
+    public void saveAllData(List<ShipHashEntity> entities) {
+        if (entities == null || entities.isEmpty()) {
+            return;
+        }
+
+        log.info("{} 전체 저장 시작: {} 건", getEntityName(), entities.size());
+
+        // INSERT와 UPDATE 분리
+        List<ShipHashEntity> toInsert = entities.stream()
+                .filter(e -> extractId(e) == null || !existsByImo(extractId(e)))
+                .toList();
+
+        List<ShipHashEntity> toUpdate = entities.stream()
+                .filter(e -> extractId(e) != null && existsByImo(extractId(e)))
+                .toList();
+
+        if (!toInsert.isEmpty()) {
+            batchInsert(toInsert);
+        }
+
+        if (!toUpdate.isEmpty()) {
+            batchUpdate(toUpdate);
+        }
+
+        log.info("{} 전체 저장 완료: 삽입={} 건, 수정={} 건", getEntityName(), toInsert.size(), toUpdate.size());
+    }
+
+    public boolean existsByImo(String imo) {
+        String sql = String.format("SELECT COUNT(*) FROM %s WHERE %s = ?", getTableName(), getIdColumnName("imo_number"));
+        Long count = jdbcTemplate.queryForObject(sql, Long.class, imo);
+        return count != null && count > 0;
+    }
+}

src/main/java/com/snp/batch/jobs/shipdetail/batch/writer/ShipDetailDataWriter.java

@@ -1,33 +1,80 @@
 package com.snp.batch.jobs.shipdetail.batch.writer;
 
 import com.snp.batch.common.batch.writer.BaseWriter;
+import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailUpdate;
+import com.snp.batch.jobs.shipdetail.batch.entity.OwnerHistoryEntity;
 import com.snp.batch.jobs.shipdetail.batch.entity.ShipDetailEntity;
+import com.snp.batch.jobs.shipdetail.batch.entity.ShipHashEntity;
 import com.snp.batch.jobs.shipdetail.batch.repository.ShipDetailRepository;
+import com.snp.batch.jobs.shipdetail.batch.repository.ShipHashRepository;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
-import java.util.List;
+import java.util.*;
+import java.util.stream.Collectors;
 
 /**
  * 선박 상세 정보 Writer
  */
 @Slf4j
 @Component
-public class ShipDetailDataWriter extends BaseWriter<ShipDetailEntity> {
+public class ShipDetailDataWriter extends BaseWriter<ShipDetailUpdate> {
 
     private final ShipDetailRepository shipDetailRepository;
+    private final ShipHashRepository shipHashRepository;
 
-    public ShipDetailDataWriter(ShipDetailRepository shipDetailRepository) {
+    public ShipDetailDataWriter(ShipDetailRepository shipDetailRepository, ShipHashRepository shipHashRepository) {
         super("ShipDetail");
         this.shipDetailRepository = shipDetailRepository;
+        this.shipHashRepository = shipHashRepository;
     }
 
     @Override
-    protected void writeItems(List<ShipDetailEntity> items) throws Exception {
+    protected void writeItems(List<ShipDetailUpdate> items) throws Exception {
+
+        if (items.isEmpty()) { return; }
+
         log.info("선박 상세 정보 데이터 저장: {} 건", items.size());
 
-        shipDetailRepository.saveAll(items);
+        // 1. List<ShipDetailUpdate> -> 3가지 List 형태로 가공
+        // 1-1. List<ShipDetailEntity> (Core20 데이터 처리용)
+        List<ShipDetailEntity> coreEntities = items.stream()
+                // ShipDetailUpdate DTO의 getShipDetailEntity() 메서드를 사용하여 바로 추출
+                .map(ShipDetailUpdate::getShipDetailEntity)
+                .collect(Collectors.toList());
+
+        // 1-2. List<List<OwnerHistoryEntity>> -> List<OwnerHistoryEntity> (OwnerHistory 데이터 처리용)
+        // OwnerHistory는 Bulk 처리를 위해 단일 평탄화된 리스트로 만듭니다.
+        List<OwnerHistoryEntity> ownerHistoriyListEntities = items.stream()
+                .flatMap(item -> {
+                    // ⚠️ ShipDetailUpdate DTO에 List<OwnerHistoryEntity> 필드가 존재해야 합니다.
+                    List<OwnerHistoryEntity> histories = item.getOwnerHistoryEntityList();
+                    return histories != null ? histories.stream() : new ArrayList<OwnerHistoryEntity>().stream();
+                })
+                .collect(Collectors.toList());
+
+        // 1-3. List<ShipHashEntity> (Hash값 데이터 처리용)
+        List<ShipHashEntity> hashEntities = items.stream()
+                .map(item -> new ShipHashEntity(item.getImoNumber(), item.getNewMasterHash()))
+                .collect(Collectors.toList());
+
+        // 2. 각 Repository에 전달
+        // ✅ 2-1. ShipDetailRepository (Core20 데이터)
+        log.debug("Core20 데이터 저장 시작: {} 건", coreEntities.size());
+        shipDetailRepository.saveAllCoreData(coreEntities);
+
+        // ✅ 2-2. OwnerHistory (OwnerHistory 데이터)
+        log.debug("OwnerHistory 데이터 저장 시작: {} 건", ownerHistoriyListEntities.size());
+        shipDetailRepository.saveAllOwnerHistoryData(ownerHistoriyListEntities);
+        
+        // TODO : 추가적인 Array/List 데이터 저장 로직 추가
+
+        // ✅ 2-3. ShipHashRepository (Hash값 데이터)
+        log.debug("Ship Hash 데이터 저장 시작: {} 건", hashEntities.size());
+        shipHashRepository.saveAllData(hashEntities);
+
+        log.info("선박 상세 정보 및 해시 데이터 저장 완료: {} 건", items.size());
 
-        log.info("선박 상세 정보 데이터 저장 완료: {} 건", items.size());
     }
+
 }

src/main/resources/application.yml

@@ -4,7 +4,7 @@ spring:
 
   # PostgreSQL Database Configuration
   datasource:
-    url: jdbc:postgresql://61.101.55.59:5432/snpdb
+    url: jdbc:postgresql://211.208.115.83:5432/snpdb
     username: snp
     password: snp#8932
     driver-class-name: org.postgresql.Driver