Hyesung Oh

서론 배치성 Spark Job은 Airflow에서 trigger 하였지만 Streaming application은 이와 달리 클러스터를 상시 점유하고 있어 다음과 같은 요구사항을 가진다. 애플리케이션 배포, 상태를 선언적으로 관리할 수 있어야 한다. 하나 이상의 Kakfa Topic을 구독하는 애플리케이션(subscriber) 배포 시에 필요한 resource 정의 (코드)를 재활용할 수 있어야 한다. 개발자는 master에 merge 하면 코드는 자동으로 배포되어야 한다. 참고로 필자는 1을 만족하기 위해 ack-emrcontainers-controller를, 2를 위해 helm chart를, 3을 위해 argocd를 사용하였지만, 이는 각자 처한 상황에 맞게 다르게 가져갈 수 있다. 또한 아래 기술들..

Structured Streaming + Kafka Integration Guide (Kafka broker version 0.10.0 or higher) 기준 내용이며, 실무에서 Kafka를 컨슈밍하는 Spark 스트리밍 어플리케이션을 운영하며 나름대로 적용하고 실험해본 옵션들에 대한 본인만의 노트인 점 참고바란다. kafka.group.id (default None) kafka consumer는 항상 하나의 consumer group에 속하게 되는데 이에 대한 식별자이다. Spark 내부적으로 스트리밍 쿼리마다 default generation 대상이며, 이를 특별히 지정하고자 하는 경우 사용한다(예를 들어 Kafka group-based authorization). 하지만 이를 사용하게 되면 bat..

EMRFS EMR의 S3 파일 읽기 쓰기와 관련된 프로토콜 집합이며, Amazon S3로 직접 일반 파일을 읽고 쓰는 데 사용하는 HDFS 구현체이다. 그리고 여기엔 다양한 기능들이 포함되는데, 그 중에서도 대표적으로 실무에서 가장 많이 이슈를 겪었던 consistent view와 s3-optimized commiter에 대해 다뤄보려 한다. EMRFS consistent view EMR에서는 file consistency를 강화하기 위해 자체적인 consistent view 메커니즘을 지원한다. EMR이 S3 또는 file system에 파일에 대한 메타정보를 DynamoDB에 관리한다. EMR 단에서 file을 create, delete를 하게 되면, 정상적으로 DynamoDB 메타정보와 sync가 ..

EMR Core node의 hdfs (disk)의 점유율이 90% (default threshold 90%) 이상이 되어 해당 node가 unhealthy 상태인 것을 확인할 수 있습니다. If the node remains unhealthy for 45 minutes, Amazon EMR marks the associated Amazon EC2 instance for termination as FAILED_BY_MASTER.When all Amazon EC2 instances associated with core nodes are marked for termination, the cluster terminates with the status NO_SLAVE_LEFT because there are no..

조금만 검색해보면 spark Performance tuning과 관련된 좋은 참고자료들이 많이 있습니다. 그 중에서 실제 팀에서 적용하여 효과를 보고있는 내용만 선별하여 공유하고자 합니다. 크게 코드 레벨에서의 최적화와 configuration 레벨에서의 최적화 두 가지가 있을 것 같습니다. 1. Code Level Opimization point 1. filter -> aggregation. aggregation을 하게 되면 driver 노드에 많은 부하가 있을 수 있습니다. 따라서 이럴 경우 reduceByKey를 이용하여 driver로 전송되는 데이터 사이즈를 최대한 줄이는게 포인트입니다. point 2. Iterator 최대한 활용하기 driver node로 데이터를 불러와서 작업을 해야하는 경우..

Pyspark로 이전하면서 느꼈던 가장 큰 아쉬움 한가지는 바로, Scala의 Dataset API가 지원되지 않는다는 점이었습니다. Dataset API를 통해 객체지향 인터페이스를 마음껏 누리면서도 Spark tunsgten project 이후 (Scala code -> java object -> 자연스레 cpu, memory 오버헤드) CPU, Memory 작업에서 많은 최적화가 이루어져 성능도 우수한 편이었기에 개인적으로 선호했기 때문입니다. Pyspark에서 이러한 아쉬움을 달래기 위해 노력한 과정과 그 과정에서 바뀌게된 생각들을 공유하고자 합니다. Code Style Guide 코드 스타일 가이드는 https://github.com/palantir/pyspark-style-guid를 우선적으로..

Spark가 JVM 위에서 동작하는 사실은 Spark 개발자라면 누구나 아는 사실입니다. Pyspark 구동의 핵심 부분인 Python process와 JVM process간의 객체 레벨 통신에 대해서 궁금증이 생겼습니다. 아래 본문은 Pyspark 소스코드를 파헤치며 파악한 내용들이며 잘못된 내용에 대한 피드백 주시면 책임감을 가지고 수정하겠습니다. Overview PySpark is built on top of Spark's Java API. Data is processed in Python and cached / shuffled in the JVM The In the Python driver program, SparkContext uses Py4J to launch a JVM and create a..

Scala Spark에서 Pyspark로의 이전을 진행하며 겪었던 시행착오와 고민들을 공유하고자 합니다. 프로젝트 구조 spark 아래에 core, tasks, util 폴더를 두었습니다. tasks: spark-submit 호출시 PY_FILE로 넘겨주는 python file이며 spark context를 초기화하며 pyspark dataframe api를 이용하여 비즈니스로직을 구현합니다. core: 모든 Spark application에서 데이터 로드, 처리, 적재시 공통적으로 사용되는 논리적인 개념을 추상화한 모듈을 정의했습니다. 아래에서 자세히 다루겠습니다. util: tasks 또는 core에서 공통적으로 사용되는 패턴 또는 utility 기능들을 정의했습니다. 1. tasks - app - ..