Hyesung Oh

TL;DR개발자가 작성하는 많은 비즈니스로직에는 특정 조건, 집단에 해당하는 item, user만 포함 or 제외하는 형태가 많은 부분을 차지합니다. 특히 데이터 엔지니어의 경우 통계 마트 테이블을 만들일이 많기 때문에 이와 같은 패턴에 익숙할 것입니다. 그중 Spark를 예를 들어, (1) user별 구매이력, (2) user 정보 두 테이블이 있을 때, user id를 key로 하여 두 테이블을 join 후 특정 조건 (1)에 해당하는 row만 (2)에서 filter 하는 식일 것입니다.하지만 join시 여러 worker node간에 분산 배치되어 있는 동일 key들을 동일 partition에 위치시키기 위해 필연적으로 동반되는 shuffling 동작은 데이터가 커질 시 병목으로 작용하게 됩니다. S..

개요 Spark에서는 JDBC api를 통해 접근할 수 있는 datasource(dbms)를 지원합니다. jdbc datasource를 사용하기 위해선 JDBC interface를 구현한 Driver class가 필요합니다. *현재 사용 중인 mysql-connector-java-8.0.23을 기준으로 작성했습니다. spark.read.jdbc option으로 driver class path를 아래와 같이 설정해주면 됩니다. driver class path: com.mysql.cj.jdbc.Driver # Read from MySQL Table df = spark.read \ .format("jdbc") \ .option("url", "jdbc:mysql://localhost:3306/emp") \ .o..

Kafka 내부 동작 원리 이해하기 (1) Replication 편에 이어서. 다시 한번 강조하면 이 책은 고승덕님의 실전 카프카 개발부터 운영까지를 스터디하며 이해한 내용을 내 나름대로 정리한 포스팅입니다. 따라서 책과 다른 내용이 있다면 그것은 본인이 실수이니 수정 코멘트 부탁드리겠고, 자세하고 정확한 내용은 책을 참고바랍니다. 컨슈머 오프셋 관리 컨슈머가 제대로 동작하려면 오프셋(Offset) 관리가 핵심이다. 어디까지 읽었는지에 대한 북마크라고 이해하면 된다. 그렇다면 오프셋 정보는 어디에서 관리될까? 카프카에 맡길 수도 있고, 컨슈머 자체적으로 HDFS compatible storage에 저장하여 관리할 수도 있다 (ex. Spark Structured Streaming). 전자 방식의 경우 컨..

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 - ..