개발/Spark

[Spark] Parquet 파일을 읽는 동안 Dataframe의 메타데이터 변경 에러

이 포스팅에서는 Parquet 파일을 읽는 동안 Dataframe의 메타데이터가 수정 되었을 경우 "REFRESH TABLE tableName"에러가 뜨는 문제를 어떻게 해결할 수 있는지 알아보겠습니다. 기존의 모든 배치에서는 동일 파티션에 대해 .mode("overwrite")를 사용하기 때문에, 프로세스가 실행되는 동안 수정/삭제된 파일을 읽으려고 시도하게 되면 메타데이터가 변경되었다는 "REFRESH TABLE tableName" 에러가 발생합니다. 이로 인해 해당 파티션이 저장 중 상태인 temporary 에서는 테이블을 사용할 수 없게 됩니다. 따라서 아래와 같이 메타데이터를 메모리 위에 올려주거나 다른 dump와 같은 경로에 저장을 하는 방법 등으로 해결할 수 있습니다. 저는 그 중 DataF..

[Spark] Spark deploy mode 설정(Cluster/Client)

Spark에서 클러스터 매니저로 YARN을 사용하고 있는데, 배포 방식에 따른 차이(리소스 할당 등) 때문에 위 글을 작성하게 되었다. 0. Cluster Mode? Manager가 뭐야? Cluster Manager란? 여러 대의 서버로 구성된 클러스터 환경에서 Spark 애플리케이션이 실행되는 컴퓨터 클러스터의 리소스(Driver와 Executor사이의 자원)를 관리하는 시스템이다. Cluster Manager는 Spark 애플리케이션의 다양한 작업 및 단계에 CPU, 메모리 및 네트워크 대역폭과 같은 리소스를 할당하고 이러한 작업의 실행을 모니터링하여 성공적으로 완료되었는지 확인하는 역할을 한다. 배포 방식은 Client mode와 Cluster mode 2가지 이다. 드라이버 프로그램이 지역적으..

[Spark DEBUG] partiton path load 오류

문제 상황 참조하는 테이블은 컬럼 기반의 데이터 저장 방식인 parquet 포맷. max partition이 25개로 설정되어 있음. path를 확인하는 시점과 데이터프레임을 load하는 시점에서의 리스트가 달라 path가 없다는 오류 발생. -> 25 제한에 따라 가장 오래 된 파티션이 삭제 됨으로 해당 파티션이 없다는 오류 발생 시도한 것들 기존 방법 (기존 코드를 그대로 따라 갔었는데, 해당 방법은 전체 메타데이터를 로드하는 방법임을 깨달음 stackoverflow) 배열로가져오고 맥스값 가져오기. collect가 한번 로드되면 목록 외로 데이터 전체가 로드 되는 줄 알았는데 아니었음. max_updatetime = spark.sql("""select max(updatetime) from db.ta..

[Spark DEBUG] Refresh for Dataframe while reading file 오류

Spark에서 파일이 존재하지 않는다는 문구와 함께 다음 오류가 발생한다. 캐시된 메타데이터 정보가 업데이트되지 않아 Spark SQL에서 새로 삽입된 데이터를 쿼리하지 못하여 작업 실행 오류가 발생한다고 한다. It is possible the underlying files have been updated. You can explicitly invalidate the cache in Spark by running 'REFRESH TABLE tableName' command in SQL or by recreating the Dataset/DataFrame involved. 테이블과 관련된 모든 캐시된 항목을 업데이트 함으로 해결할 수 있다고 한다. spark.sql('refresh TABLE db명.ta..