티스토리

🐥

검색하기

블로그 홈

🐥

quackstudy.tistory.com/m

•8• 님의 블로그입니다.

구독자: 1

방명록 방문하기

주요 글 목록

Spark, Trino on kubernetes Spark on KubernetesSpark에서는 cluster manager로 kubernetes를 지원한다(2.3 버전부터). kubernetes가 관리하는 클러스터에서 spark가 실행되고, 네이티브 kubernetes 스케줄러를 사용한다. 장점컨테이너 기반 워크로드쿠버네티스의 HPA를 사용해 executor 수를 자덩으로 조정할 수 있다.작업 및 배포 관리 유현성단점초기 설정 복잡성kubernetes 스케줄링 문제: third party scheduler를 사용해야 함kubernetes 클러스터의 상태에 spark 성능 및 가용성에 영향을 받음복잡한 모니터링 및 로깅 - spark 자체 로그, pod 로그, 시스템 로그, 이벤트 로그 등...data locality - kubernetes에서는 s.. 공감수 0 댓글수 0 2024. 10. 9.
중앙 집중식 로깅 시스템 (EFK stack) 중앙 집중식 로깅일반적으로 분산환경에서 마이크로서비스를 운영하면 각 서비스에서 발생한 로그 파일들은 파편화되어 각각 저장된다.서로 다른 개별 서비스에서 발생하는 로그를 연결지어 처음부터 끝까지 순서대로 추적하는 것은 어렵다.따라서 로그의 출처에 상관없이 모든 로그를 중앙 집중적으로 저장, 분석해야 하며, 서비스 실행 환경에서 분리하여 관리되어야 한다. 장점여러 서버나 어플리케이션에서 생성된 로그를 한 곳에 수집하기 때문에 다양한 시스템의 상태를 한 번에 파악할 수 있다.모든 로그 데이터를 통합하여 저장하므로 관리가 용이해진다. (보통 용량이 크고 확장 가능한 DB나 클라우드 스토리지사용)중앙에서 로그를 한 번에 검색하고 분석할 수 있으므로 이벤트를 추적하기가 쉬워지고, 문제 해결 시간이 단축된다.단점단일.. 공감수 5 댓글수 3 2024. 10. 9.
[Apache Nifi] Mysql DB 연결하기 1. Mysql 연결 jdbc 다운로드https://dev.mysql.com/downloads/connector/j/홈페이지에서 환경에 맞는 jdbc를 다운로드 받아 압축파일 내에 있는 .jar 파일을 Nifi 실행 서버에 위치시킨다.여러 노드로 클러스터링되어 있다면 모두 동일 위치에 넣어주어야 한다. 2. DBCPConnectionPool 생성Nifi > 우클릭 > Configure > Controller Services에서 DBCPConnectionPool 추가 후 설정에서 아래의 항목을 채워주면 된다.Database Connection URL: jdbc:mysql://주소:port/db명Database Driver Class Name: com.mysql.jdbc.DriverDatabase Driv.. 공감수 0 댓글수 0 2024. 7. 15.
Apache Iceberg 사용 사례 (LINE Data Platform에서 Apache Iceberg 도입) https://www.youtube.com/watch?v=7y9gNwqLNtU위 동영상 정리함 hive -> iceberg로 변경한 사례 Line 데이터 플랫폼의 기존 방식데이터 플랫폼의 특징데이터의 크기/종류/활용 규모가 크다머신 수: 5000 대+데이터 크기: 290 PB+하이브 테이블 수: 4만개 (모두 external table)log ingestion rate: 초당 17.5M+하루에 구동되는 job 수: 15만+플랫폼 사용자 수: 700+ Query ProcessingSQL 기반, spark, hive, trino, flink를 분산 sql 쿼리 엔진으로 사용데이터파일이 어디에 있는지, 어떻게 읽고 쓸수있는지에 대한 정보 필요쿼리에 대한 테이블을 어떻게 관리할지는 테이블 포맷에 의해 정의됨 *.. 공감수 0 댓글수 0 2024. 5. 23.
[Spark] 스파크 스트리밍의 이벤트 시간 처리와 Watermark 스트리밍 쿼리에서의 집계는 보통 현재시간부터 T시간 전까지 집계된 데이터를 의미한다. 그런데 예를들어 지난 15초 동안의 집계 데이터라고 하면 15초의 시간은 언제부터 언제까지인지 의문이 생길 수 있다. 스트림 처리에서 바라보는 시간은 다양한 기준이 있다. 스트림 처리에서의 시간 관점 이벤트 시간 (event time) 데이터 의존적인 타임스템프로, 데이터 내에 존재하는 데이터 발생 시간이다. 데이터 의존적이기 때문에 어떠한 값을 타임스탬프로 넣을 것인지는 다르겠지만 주로 이벤트가 발생한 시간을 많이 사용한다. windowing 에 사용할 수 있는 아주 적절한 시간 중 하나이다. 수집 시간 (ingestion time) 스트림 처리 엔진에 데이터가 처음으로 수집된, 즉 들어온 시간을 나타낸다. 처리 시간.. 공감수 0 댓글수 0 2024. 4. 21.
[Spark] Structured Streaming - stateful transformation과 Window operation streaming 코드를 작성하면서 filter, count 등 여러가지 연산자를 사용할 수 있다.과거 데이터의 정보가 필요없는 stateless 연산자와, 과거의 데이터가 필요해 데이터 유지가 필요한 statefull 연산자로 나눌 수 있다.Stateless Trasformation과거의 데이터와 상관 없이 현재의 데이터만 사용하는 연산자Stateful Transfmraion이전 배치의 데이터의 유지가 필요한 연산자예: aggreagion, counter, join, group, windowing 등Stateless Transformation각 배치를 이전 배치를 참조하지 않고 독립적으로 처리한다. -> 배치의 출력이 해당 배치의 데이터만을 기반으로 함을 의미select, filter, map, fla.. 공감수 1 댓글수 0 2024. 4. 16.
[Kubernetes] NodeSelector 설정 deployment 나 statefulsets yaml 파일에서 설정하는 방법 `spec.template.spec.nodeSelector`추가 (containers 와 같은 depth) 추가로 데이터타입이 map[string] 이어야해서 숫자로만 이루어져있으면 따옴표 추가 필요 공감수 0 댓글수 0 2024. 4. 9.
[Airflow] Nifi Trigger Operator 만들기 airflow에서 nifi processor로 trigger를 보내야 했다. 사실 nifi에서 restAPI를 지원해주기 때문에 NIFI ETL 자체를 시작하는 것은 큰 문제가 안됐지만 nifi 작업 이후에 airflow에서 후속 task가 있다면 ETL이 어느 시점에 끝나는 지를 확인하는 게 어려웠다. NIFI는 processor 단위로 데이터를 처리하며, flowfile이 processor에 인풋되면 바로 처리 후 output stream으로 내보내기 때문에 종료 시점을 명확히 알 수가 없었다. 아래 블로그에서 airflow에서 nifi 를 트리거할 수 있는 방법을 소개했지만 시작 노드를 `GenerateFlowFile`로 설정해야 하는 등 제약조건이 있었다. https://towardsdatasci.. 공감수 0 댓글수 0 2024. 4. 4.
[Spark] Spark Structured Streaming - Fault Tolerance Background 실시간 stream 처리는 지속적인 input stream의 특성 상 중단되지 않고 24시간 실행되므로 여러 오류 원인으로 failure이 발생할 수 있다. input stream이 작성된 코드로 처리될 수 없는 경우 (잘못된 형식의 데이터) 시스템/cluster 오류의 경우 Spark Streaming에서의 Fault Tolerance 정의 스파크의 목표는 end-to-end Exactly Once 보장이다. (참고) At Most Once: 각 레코드는 한 번만 처리되거나 아예 처리되지 않는다. At Least Once: 각 레코드가 한 번 이상 처리된다. 데이터가 손실되지 않도록 보장하므로 At Most Once보다 강력하지만 중복이 있을 수 있다. Exactly Once: 각 .. 공감수 0 댓글수 0 2024. 4. 4.
[Spark] Spark Structured Streaming 개요 Spark Streaming 이란 core spark API의 확장 프로그램으로 분산 스트림 처리 프로세싱을 지원한다. streaming 타입으로는 아래와 같이 두 종류가 있는데 spark streaming은 RDD 베이스 엔진으로, 2.x버전까지 지원하고 이후 더이상 업데이트 되지 않는 레거시 프로젝트이다. Spark Streaming 종류 Spark Streaming: RDD 기반의 micro-batch 수행 Spark Structured Streaming: Dataframe 기반 micro-batch 수행, 저지연 처리 모드를 도입함으로써 실시간에 가까운 처리가 가능해짐. Spark Structured Streaming 프로그래밍 모델 structured streaming에서는 트리거 간격마다 수신.. 공감수 0 댓글수 0 2024. 4. 4.
[Spark] Accumulator와 Broadcast (공유변수) 참고: https://spark.apache.org/docs/latest/rdd-programming-guide.html#shared-variables 공유변수 스파크에서는 분산하여 병렬처리하기 위해 데이터를 분할해서(파티셔닝) 여러 머신에서 동시에 처리한다. 이때 사용된 모든 변수는 배포될 때 복사된 복사본으로 작동하는데, 이러한 변수의 업데이트는 driver에 다시 전달되지 않는다. 그러나 어떤 조건(?) 을 가지고 있어야 데이터를 처리할 수 있는 경우가 발생하는데, 이럴때 제한된 유형의 공유 변수를 사용하여 병렬 처리 후 원하는 값을 받을 수 있다. Broadcast Variable 브로드캐스트 변수를 사용하면 변수의 복사본을 전달하는게 아닌, Read Only 변수를 각 워커 노드에 캐시된 상태로.. 공감수 0 댓글수 0 2024. 4. 1.
[Spark] SQL Hint SQL Hint 참고: https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-hints.html 문법 /*+ hint [ , ... ] */ 파티셔닝 힌트 파티셔닝 전략을 제안할 수 있다. COALESCE `coalesce` dataset API와 동일하다. 매개변수는 파티션 개수이고, 지정된 파티션 수로 줄일 수 있다. SELECT /*+ COALESCE(3) */ * FROM t; REPARTITION `repartition` dataset API 와 동일하다.파티션 개수, 컬럼명을 매개변수로 사용할 수 있고 지정된 수의 파티션으로 다시 분할하는데에 사용된다. SELECT /*+ REPARTITION(3) */ * FROM t; SELECT .. 공감수 0 댓글수 0 2024. 4. 1.
[Spark] 스파크 스케쥴링 Application 스케쥴링 관련하여 https://community.cloudera.com/t5/Community-Articles/Dynamic-Allocation-in-Apache-Spark/ta-p/368095 이쪽에 정리가 잘 되어 있다. SRA(Static Resource Allocation) spark 어플리케이션은 실행되기 전 어플리케이션에서 사용할 리소스를 미리 예약해둔다. 이 리소스의 양은 고정되어서 런타임중에는 변경이 불가능하다. 만약 할당된 것보다 더 많은 리소스가 필요하다면 실행시간이 길어지거나 실패할 수 있다. 리소스가 고정되어 있기 때문에 만약 할당된 리소스가 어플리케이션에서 사용되지 않는다면 비효율적인 리소스 활용으로 이어지게 된다. 또한 런타임중에 리소스를 조정하는것이 불가.. 공감수 0 댓글수 0 2024. 4. 1.
[Spark] Repartition과 Coalesce spark를 통해 연산을 하다 보면 파티션이 과다하게 많거나 너무 적은 상황이 있을 수 있다. 이럴 때에는 repartition과 coalesce를 통해 파티션 개수를 지정해줄 수 있다. `repartition()`과 `coalesce()` 모두 파티션 개수를 설정해주는 함수이다. 파티션의 개수를 조정해주지만 `coalesce()`는 파티션의 개수를 줄이는 것만 가능하다는 점에서 차이점이 있다. 1. 실행 예시 repartition과 coalesce의 가장 큰 차이점은 shuffle 수행 여부에 있다. `coalesce`를 사용해서 파티션 수를 줄인 경우, stage2에서 coalesce 가 수행되는 것을 확인할 수 있다. load df1 load df2 df3 = df1.join(df2, some_co.. 공감수 0 댓글수 0 2024. 4. 1.
[Spark] cache()와 persist() 스파크로 데이터를 처리하면 action 연산자 수행 시에 데이터가 로드된다. 반복된 동일한 연산을 한다면 매번 action 때 로드하지 않고 cache()와 persist()를 사용하여 데이터를 메모리에 상주시킬 수 있다. 캐시된 데이터는 storage memory에 보관된다. 1. cache() 와 persist() cache와 persist 모두 동일한데 persist에서는 storageLevel을 설정해줄 수 있다는 점에서 차이가 있다. `cache()`는 Dataframe 의 경우`persist(storageLevel=MEMORY_AND_DISK)` , RDD의 경우 ` persist(storageLevel=MEMORY_ONLY)` 와 같은 함수이다. https://spark.apache.org/.. 공감수 0 댓글수 0 2024. 4. 1.
[Spark] Speculative Execution 분산 환경에서는 고르지 못한 리소스, 데이터 로드 등 task 실행 속도가 느려지는 상황이 많이 있다. Speculative Execution 한 단계에서 여러 작업의 실행 시간이 다를 수 있다. 다른 task와 비교하여 특히 느리게 실행되는 Task가 있을 수 있는데, 클러스터 내 노드의 구성 성능 차이, 네트워크 변동 등의 이유가 있을 수 있다. speculative execution은 동일한 단계에 오랫동안 완료되지 않는 작업이 있을 경우, 다른 노드에서 동일한 작업을 시작하고, 먼저 완료되는 task를 성공처리하고, 이외의 다른 시도는 종료한다. 클러스터 내 노드 간 성능 차이가 있는 경우 speculative execution 기능을 활성화하는 것이 좋다. 주요 구성 옵션 spark.specul.. 공감수 0 댓글수 0 2024. 4. 1.
[Airflow] Telegram 알림 operator 만들기 (requests 사용) 버전정보 Airflow 2.5.3 Python 3.8 왜인지는 모르겠지만 2.5.3 버전을 사용하고 있는 환경에서는 telegram provider 설치는 되는데 connection type 리스트에는 안떴다 ㅠ (버전호환성도 다 맞춤) 2.6.3 버전으로 올리고 provider 재설치 후 다시 확인해보니 operator가 잘 동작하는 건 확인했는데.. 어쨌든 2.5.3 버전에서 사용하기 위해 request 라이브러리를 이용해서 post 요청을 보내는 방식으로 Webhook을 만들었다. 1. Telegram Connection 생성 이건 그냥 hook에서 connection 정보 어떻게 가져올지 상황에 맞게 설정하면 될 듯 하다. 나같은 경우는 그냥 http connection type을 사용했다. `H.. 공감수 0 댓글수 0 2024. 3. 29.
[Spark] Logical Plan 과 Physical Plan 보호되어 있는 글입니다. 공감수 0 댓글수 0 2024. 3. 25.
[Spark] RDD vs Dataframe 스파크에서 데이터 구조 종류에는 세 개가 있다. RDD: spark 1.0 도입 Dataframe: spark 1.3 도입 Dataset: spark 1.6 도입 참고로 spark 2.0부터 dataframe api는 datasets과 통합되었다. (스칼라와 자바에서) ` Unifying DataFrames and Datasets in Scala/Java: Starting in Spark 2.0, DataFrame is just a type alias for Dataset of Row. ` 1. RDD란? Reillient Distributed Data로 풀어쓰여지는 RDD는 회복력 있는/불변의 분산 데이터 정도로 해석될 수 있다. 스파크의 가장 기본적인 데이터 단위이다. 불변의 특성 RDD는 이름(Re.. 공감수 1 댓글수 2 2024. 3. 24.
[Spark] spark-submit 과 옵션 spark-submit 이란? spark-submit은 스파크 application을 클러스터에 배포할 때 사용하는 스크립트이다. 이 스크립트는 application에서 사용하는 리소스를 제어하는 여러 플래그를 제공한다. spark-submit 옵션 ./bin/spark-submit \ --class \ --master \ --deploy-mode \ --conf = \ ... # other options \ [application-arguments] --py-files file1.py,file2.py,file3.zip my_python_file.py 위의 spark-submit 주요 옵션 예시의 플래그 값들을 하나씩 살펴보면 아래와 같다. master 플래그 (cluster manager) spark .. 공감수 0 댓글수 1 2024. 3. 24.
[Spark] Adaptive Query Execution(AQE) Spark 2.x 에서는 rule-based/cost-based의 방식으로 쿼리를 최적화한다. Spark 3.0부터는 런타임에 최적화할 수 있는 AQE가 도입되었고, Spark 3.2 부터는 AQE 활성화가 디폴트 버전이다. AQE에서는 쿼리를 최적화하기 위해 크게 아래의 세 가지의 기능을 지원한다. Coalescing Post shuffle partitions Switching join strategies Optimizing Skew Join Coalescing Shuffle Partitions 셔플 파티션의 수를 최적화해주는 기능이다. 셔플은 스파크 어플리케이션의 성능에 매우 큰 역할을 차지하고, 기존에는 사용자가 셔플 크기 및 파티션을 직접 확인하며 수동으로 파티션 수를 조정해주어야 했다. AQE에.. 공감수 0 댓글수 0 2024. 3. 23.
[Spark] 스파크의 Executor Memory 구조 참고: https://community.cloudera.com/t5/Community-Articles/Spark-Memory-Management/ta-p/317794 Spark Executor Memory 구조 Executor Container의 메모리 구조는 이미지와 같이 크게 Memory Inside of JVM: spark.executor.memory Memory Outside of JVM: spark.yarn.executor.MemroyOverHead 로 나뉜다. Object의 읽기/쓰기 속도는 `on-heap > off-heap > disk` 순서로 빠르다. Memory Inside of JVM (=On-Heap Memory = In-Memory) java GC 관리 하에 있는 메모리 영역이다... 공감수 0 댓글수 0 2024. 3. 23.
[Spark] GraphX graphX는 스파크에서 graph computation을 수행하기 위한 Spark API이다. 스파크 문서에 설명되어 있는 GraphX의 장점은 아래와 같다. Flexibility 단일 시스템 내에서 ETA, 분석, 그래프 계산을 통합할 수 있다. RDD/Dataframe에서 그래프로 효율적인 변환이 가능하며, Pregel API로 사용자 정의 반복 그래프 알고리즘을 작성할 수 있다. Speed 실제 그래프 컴퓨팅 목적으로 만들어진 시스템 (specialized graph processing system)과 비교하여 비슷한 성능을 보인다. Algorithm 다양한 알고리즘이 있다. PageRank Connected components Label propagation SVD++ Strongly conne.. 공감수 0 댓글수 0 2024. 3. 18.
[Spark] Spark Join 종류 스파크에서 Join 시에는 상황에 따라 Shuffle 연산을 수행해야 하고, 이는 executor 사이의 방대한 데이터 이동을 야기한다. 스파크에서 사용되는 조인의 방식은 아래와 같다. 1. Broadcast Hash Join 두 개의 데이터를 조인할 때 한 쪽이 매우 작고 한 쪽은 매우 큰 사이즈의 데이터일 때, 더 작은 쪽의 데이터가 Driver에 의해 모든 executor로 복사되는 방식이다. 사이즈가 작은 테이블의 키 값을 해시 테이블로 만든다. 해시 테이블의 값을 큰 테이블 파티션에 복사해준다. 각 파티션 내부에서 조인을 진행한다. 이 방식은 shuffle을 생략할 수 있기 때문에 데이터 이동에 따른 코스트가 사라진다. 모든 executor에 데이터를 복사하기 때문에 복사하려는 dataset이 .. 공감수 1 댓글수 0 2024. 3. 18.
인터넷 안되는 환경에서 yum 패키지 설치하기 준비물: 인터넷이 되는 vm, rpm 1. rpm 패키지 다운로드 yum install --downloadonly --downloaddir=$다운로드받을경로 $패키지명 # 예시 yum install --downloadonly --downloaddir=./test librdkafka-devel-0.11.4-1.el7.x86_64 패키지 명은 정확하게 적어줘야한다. 이렇게 사용하는 이유는 의존패키지까지 알아서 다운로드 받아주기 때문이다. 예를들어 `librdkafka-devel`의 경우에는 아래와 같이 두 개의 rpm 파일이 받아진다. (디펜던시때문에) librdkafka-0.11.4-1.el7.x86_64.rpm librdkafka-devel-0.11.4-1.el7.x86_64.rpm 2. 설치하려는 서버.. 공감수 0 댓글수 0 2024. 3. 12.
[Spark] spark에서 s3 접근하기 (ACCESS_KEY, SECRET_KEY) 2.4.4 이하와 2.4.5 이상 버전에서 하둡 configuration 설정하는게 좀 다른 듯 하다. Spark 버전 2.4.4 이하 spark = SparkSession.builder.appName("myapp") \ .config("some.config", "some.value") \ .getOrCreate() # signature V4 를 사용하는 리전일 경우 아래 property 설정 spark.sparkContext.setSystemProperty("com.amazonaws.services.s3.enableV4", "true") # s3 정보 설정 spark.sparkContext._jsc.hadoopConfiguration().set(f"fs.s3a.bucket.{mybucket}.endpo.. 공감수 0 댓글수 0 2023. 12. 19.
파일 수정 시간 초단위 확인 ls --time-style='+%Y%m%d %H:%M:%S' -l 공감수 0 댓글수 0 2023. 10. 12.
[Kafka] kafka produce 하기 (with C) 처음엔 그냥 파이썬으로 파일을 읽어서 카프카 메시지를 보내려 했는데 생각보다 느렸다.. 카프카로 보내야하는 파일을 생성하는 코드가 c/c++로 되어있기도 하고 메시지 생성 즉시 바로 보내면 더 빨라질 것 같아 C로 작성해봤다. 1. librdkafka 패키지 다운로드 나는 centos 써서 아래와 같이 설치했다. yum install librdkafka-devel 다른 운영체제는 아래 github 페이지 참고 https://github.com/confluentinc/librdkafka#installation 2. librdkafka 설치 위치 확인 설치 위치 확인하는 방법 참고: https://quackstudy.tistory.com/entry/CentOS-yum%EC%9C%BC%EB%A1%9C-%EC%.. 공감수 0 댓글수 0 2023. 8. 31.
[CentOS] yum으로 설치한 패키지 설치 경로 확인하는 방법 yum 패키지 설치 리스트 확인 yum install list rpm 으로 경로 찾기 경로를 찾으려면 yum만으로는 안되고 rpm을 사용해야 한다. # 전체 설치 리스트 확인 rpm -qa # 특정 문자열이 들어간 패키지 이름 확인 rpm -qa | grep $문자열 # 패키지 이름으로 위치 찾기 rpm -ql $패키지명 예시 아래와 같이 설치 경로를 확인할 수 있다. [root@testserver ~]# rpm -ql librdkafka-devel-0.11.4-1.el7.x86_64 /usr/include/librdkafka /usr/include/librdkafka/rdkafka.h /usr/include/librdkafka/rdkafkacpp.h /usr/lib64/librdkafka++.so /u.. 공감수 0 댓글수 0 2023. 8. 28.
[airflow] mariadb를 external database로 사용 시 데이터베이스 관련 오류 (column doesn't exist) 정확히 어떤 오류 문구였는지는 db 옵션값을 이미 수정해버려서 재현은 어렵지만 airflow 스케쥴러에서 db를 쿼리할 때 syntax error가 난다는 오류였다. (대충 아래 오류 느낌..) db init을 하는데 select 절에 있는 컬럼들이 group by 절에 없는 오류 "table.column" doesn't exist. airflow에서는 공식적으로 mariadb를 지원하지 않는다. (참고: https://airflow.apache.org/docs/apache-airflow/stable/howto/set-up-database.html) Despite big similarities between MariaDB and MySQL, we DO NOT support MariaDB as a back.. 공감수 0 댓글수 0 2023. 8. 14.

문의안내

티스토리
로그인
고객센터

티스토리는 카카오에서 사랑을 담아 만듭니다.