diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/v2/ranking/application/CreatorRankingSnapshotJobService.kt b/src/main/kotlin/kr/co/vividnext/sodalive/v2/ranking/application/CreatorRankingSnapshotJobService.kt
index 33e48e38..d503d30e 100644
--- a/src/main/kotlin/kr/co/vividnext/sodalive/v2/ranking/application/CreatorRankingSnapshotJobService.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/ranking/application/CreatorRankingSnapshotJobService.kt
@@ -5,6 +5,7 @@ import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobPor
 import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobRecord
 import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobStatus
 import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobTrigger
+import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Service
 import org.springframework.transaction.annotation.Transactional
 import java.time.LocalDateTime
@@ -17,6 +18,7 @@ class CreatorRankingSnapshotJobService(
     private val jobPort: CreatorRankingSnapshotJobPort,
     private val nowProvider: () -> ZonedDateTime = { ZonedDateTime.now() }
 ) {
+    private val log = LoggerFactory.getLogger(javaClass)
     private val periodPolicy = CreatorRankingPeriodPolicy()
 
     @Transactional
@@ -37,11 +39,14 @@ class CreatorRankingSnapshotJobService(
         )
         val jobId = job.id ?: return
         jobPort.markProcessing(jobId, LocalDateTime.now())
+        logJobStatusChanged(job, CreatorRankingSnapshotJobStatus.PROCESSING)
         try {
             refreshService.refreshLastCompletedWeek(now)
             jobPort.markDone(jobId, LocalDateTime.now())
+            logJobStatusChanged(job, CreatorRankingSnapshotJobStatus.DONE)
         } catch (ex: Exception) {
             jobPort.markFailed(jobId, LocalDateTime.now(), ex.message)
+            logJobStatusChanged(job, CreatorRankingSnapshotJobStatus.FAILED, ex.message)
             throw ex
         }
     }
@@ -83,4 +88,21 @@ class CreatorRankingSnapshotJobService(
 
         jobPort.markPending(jobId)
     }
+
+    private fun logJobStatusChanged(
+        job: CreatorRankingSnapshotJobRecord,
+        status: CreatorRankingSnapshotJobStatus,
+        error: String? = null
+    ) {
+        log.info(
+            "event=creator_ranking_snapshot_job_status_changed " +
+                "jobId={} trigger={} status={} aggregationStartAtUtc={} aggregationEndAtUtc={} error={}",
+            job.id,
+            job.trigger,
+            status,
+            job.aggregationStartAtUtc,
+            job.aggregationEndAtUtc,
+            error
+        )
+    }
 }
diff --git a/src/test/kotlin/kr/co/vividnext/sodalive/v2/ranking/application/CreatorRankingSnapshotJobServiceTest.kt b/src/test/kotlin/kr/co/vividnext/sodalive/v2/ranking/application/CreatorRankingSnapshotJobServiceTest.kt
index 52fc6d10..77942094 100644
--- a/src/test/kotlin/kr/co/vividnext/sodalive/v2/ranking/application/CreatorRankingSnapshotJobServiceTest.kt
+++ b/src/test/kotlin/kr/co/vividnext/sodalive/v2/ranking/application/CreatorRankingSnapshotJobServiceTest.kt
@@ -6,13 +6,18 @@ import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobSta
 import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobTrigger
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertThrows
+import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.DisplayName
 import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.ExtendWith
 import org.mockito.Mockito
+import org.springframework.boot.test.system.CapturedOutput
+import org.springframework.boot.test.system.OutputCaptureExtension
 import java.time.LocalDateTime
 import java.time.ZoneId
 import java.time.ZonedDateTime
 
+@ExtendWith(OutputCaptureExtension::class)
 class CreatorRankingSnapshotJobServiceTest {
     @Test
     @DisplayName("스케줄 실행은 집계 기간을 포함한 SCHEDULED job을 생성하고 성공 시 DONE으로 기록한다")
@@ -154,6 +159,44 @@ class CreatorRankingSnapshotJobServiceTest {
         assertEquals(CreatorRankingSnapshotJobStatus.PENDING, unchanged.status)
         assertEquals("keep", unchanged.lastError)
     }
+
+    @Test
+    @DisplayName("스케줄 job 상태 변경은 job id와 상태를 로그로 남긴다")
+    fun shouldLogScheduledJobStatusTransitions(output: CapturedOutput) {
+        val refreshService = Mockito.mock(CreatorRankingSnapshotRefreshService::class.java)
+        val jobPort = FakeCreatorRankingSnapshotJobPort()
+        val now = ZonedDateTime.of(2026, 6, 8, 7, 30, 0, 0, ZoneId.of("Asia/Seoul"))
+        val service = CreatorRankingSnapshotJobService(refreshService, jobPort) { now }
+
+        service.refreshLastCompletedWeekByScheduledJob()
+
+        assertTrue(output.out.contains("event=creator_ranking_snapshot_job_status_changed"))
+        assertTrue(output.out.contains("jobId=1"))
+        assertTrue(output.out.contains("trigger=SCHEDULED"))
+        assertTrue(output.out.contains("status=PROCESSING"))
+        assertTrue(output.out.contains("status=DONE"))
+    }
+
+    @Test
+    @DisplayName("실패 job 상태 변경은 실패 상태와 사유를 로그로 남긴다")
+    fun shouldLogFailedScheduledJobStatusTransition(output: CapturedOutput) {
+        val refreshService = Mockito.mock(CreatorRankingSnapshotRefreshService::class.java)
+        val jobPort = FakeCreatorRankingSnapshotJobPort()
+        val now = ZonedDateTime.of(2026, 6, 8, 7, 30, 0, 0, ZoneId.of("Asia/Seoul"))
+        val service = CreatorRankingSnapshotJobService(refreshService, jobPort) { now }
+        Mockito.doThrow(IllegalStateException("aggregate failed"))
+            .`when`(refreshService).refreshLastCompletedWeek(now)
+
+        assertThrows(IllegalStateException::class.java) {
+            service.refreshLastCompletedWeekByScheduledJob()
+        }
+
+        assertTrue(output.out.contains("event=creator_ranking_snapshot_job_status_changed"))
+        assertTrue(output.out.contains("jobId=1"))
+        assertTrue(output.out.contains("trigger=SCHEDULED"))
+        assertTrue(output.out.contains("status=FAILED"))
+        assertTrue(output.out.contains("error=aggregate failed"))
+    }
 }
 
 private class FakeCreatorRankingSnapshotJobPort : CreatorRankingSnapshotJobPort {