Databricks 노트북에서 데이터 품질 체크(에러 레코드 수), 간단한 카운팅/합산, 디버깅용 메트릭을 분산 실행(Executors) 중에 모으고 싶을 때가 있습니다. 이때 가장 먼저 떠오르는 게 Spark의 Accumulator(누적기) 인데요.
하지만 “돌아가긴 하는데 숫자가 가끔 이상하다”는 경험도 흔합니다. 이 글에서는 Databricks 환경에서 Accumulator의 개념 → 사용법 → 주의사항(중요) → 대안 순으로 정리합니다.
Spark의 Driver-Executor 구조 이해가 Accumulator 이해의 출발점입니다. Source
목차
- 1. Accumulator란?
- 2. Databricks 노트북에서 기본 사용법 (PySpark)
- 3. “왜 값이 이상해지지?”: 신뢰성 함정(재시도/재계산)
- 4. Broadcast vs Accumulator (헷갈림 정리)
- 5. 베스트 프랙티스 + 대안
- 참고자료
1. Accumulator란?
Spark Accumulator는 여러 Executor에서 발생한 값을 “더하기(add)” 형태로 Driver에 모으는 공유 변수입니다. 핵심 제약은 다음 한 줄로 요약됩니다.
- Executor(작업 태스크):
+=또는.add()로 “추가”만 가능 - Driver: 최종 값 조회는
acc.value로 Driver만 가능
즉, “분산 작업에서 값을 모으되, 읽기는 Driver만” 허용하는 구조입니다. Source
Databricks의 PySpark API 문서도 동일하게 설명합니다. Source
“Worker에서 누적 → Driver에서 확인”이라는 방향성이 포인트. Source
2. Databricks 노트북에서 기본 사용법 (PySpark)
Databricks 노트북에서는 보통 spark가 기본 제공되므로, sc = spark.sparkContext로 접근합니다.
2.1 가장 단순한 예시
sc = spark.sparkContext
acc = sc.accumulator(0) # 초기값 0
rdd = sc.parallelize([1, 2, 3])
def f(x):
global acc
acc += x
rdd.foreach(f)
acc.value이 형태는 Spark/Databricks 문서 예시와 동일한 흐름입니다. Source
2.2 .add()로 누적하기
b = sc.accumulator(0)
def g(x):
b.add(x)
rdd.foreach(g)
b.value이 또한 공식 예시에 포함된 패턴입니다. Source
2.3 “이건 안 됩니다”: Executor에서 value 읽기/쓰기
아래 코드는 에러가 나거나(대표적으로 Py4JJavaError), 의도한 대로 동작하지 않습니다.
- RDD 변환에서
acc.value를 읽기:
rdd.map(lambda x: acc.value).collect() # 실패(문서 예시에서 에러)- Executor에서
acc.value를 직접 설정:
def h(x):
global acc
acc.value = 7
rdd.foreach(h) # 실패(문서 예시에서 에러)이 제약은 문서에 예제로 명시되어 있습니다. Source
3. “왜 값이 이상해지지?”: 신뢰성 함정(재시도/재계산)
여기부터가 실전입니다. Accumulator는 디버깅/관측용으로는 유용하지만, 아래 상황에서 과대 카운트(over-count) 같은 문제가 생길 수 있습니다.
- 태스크/스테이지 재시도(노드 장애, 셔플 파일 유실 등)
- 셔플 재생성 때문에 이전 스테이지 태스크가 여러 번 실행
- “좀비 태스크(zombie tasks)”처럼 재시도 중에도 완료된 태스크의 업데이트가 반영되는 경우
이런 이유로 “실패 복구 과정에서 같은 작업이 여러 번 실행되며 누적이 중복 반영될 수 있다”는 비판이 잘 알려져 있습니다. Source
결론적으로, 품질검사 실패 기준(예: 에러 50개 넘으면 잡 실패) 같은 “정확해야 하는 로직”을 Accumulator 값 하나에 걸면 위험해질 수 있습니다. (특히 셔플/재시도 가능성이 있는 작업)
4. Broadcast vs Accumulator (헷갈림 정리)
Databricks 커뮤니티 글에서도 둘을 함께 소개합니다. 요약하면:
- Accumulator: Worker → Driver로 값 누적 (counter/sum)
- Broadcast: Driver → Worker로 읽기 전용 데이터 배포 (셔플/전송 반복을 줄임)
Databricks 커뮤니티 글의 설명도 같은 방향입니다. Source
| 구분 | 목적 | 데이터 흐름 | 수정 가능? | 대표 사례 |
|---|---|---|---|---|
| Accumulator | 분산 태스크의 값 “모으기” | Executor → Driver | Executor는 add만, Driver가 조회 | 에러 레코드 수, 스킵 건수 |
| Broadcast | 큰 읽기전용 값 “뿌리기” | Driver → Executor | 읽기 전용 | 룩업 테이블, 설정값 |
Databricks에서도 Spark 실행 모델(Driver/Worker) 위에서 Accumulator가 동작합니다. Source
5. 베스트 프랙티스 + 대안
5.1 Accumulator를 써도 좋은 경우
- 디버깅/관측용 메트릭(“대략 이 정도 에러가 발생하네?”)
- 로깅/모니터링에 참고할 카운터
- 정확도보다 가시성이 중요한 상황
5.2 피해야 할 경우(중요)
- 결과가 정확해야 하는 비즈니스 로직(정산/과금/정확한 오류 건수 기반 실패 처리 등)
- 셔플이 많고 장애/재시도 가능성이 높은 배치 파이프라인
- 동일 RDD/연산이 재사용/재계산될 수 있는 구조(캐시/라인리지 영향)
재시도/재계산에서 누적이 예상치 못하게 증가할 수 있다는 점은 강하게 경고됩니다. Source
5.3 대안(정확한 카운팅이 필요할 때)
- DataFrame 집계로 정식 계산:
df.filter(...).count(),groupBy().agg(...) - “나쁜 레코드”를 별도 DataFrame으로 분리 후 저장/집계(예: Delta 테이블에 quarantine)
- 품질 규칙은 Great Expectations/Deequ 같은 품질 프레임워크(조직 표준이 있다면)
참고자료
- PySpark Accumulator (Spark 공식 문서): https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.Accumulator.html
- SparkContext.accumulator 생성 API: https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.SparkContext.accumulator.html
- Databricks PySpark Accumulator API: https://api-docs.databricks.com/python/pyspark/latest/api/pyspark.Accumulator.html
- Databricks Community(Accumulator vs Broadcast): https://community.databricks.com/t5/community-articles/what-is-accumulators-and-broadcast-variables/td-p/76589
- Accumulator 신뢰성 비판/함정(재시도, 좀비 태스크 등): http://imranrashid.com/posts/Spark-Accumulators/
