Data Engineering

검색 엔진으로 알려져 있는 OpenSearch (ElasticSearch) 를 RAG로 활용한 경험에 대해 소개해보려고 한다. OpenSearch 란 무엇인가 ? ElasticSearch 와 무슨 차이지? OpenSearch 는 ElasticSearch 라는 오픈소스 제품을 AWS가 Fork 해서 개발한 제품이다. 그리고 AWS가 ElasticSearch 를 제품화 해서 서비스로 제공하는 검색 엔진 서비스가 OpenSearch 이다. 원래 오픈소스를 복사하는 일 자체는 종종 일어나지만, 오픈소스를 제품화 해서 다른 업계에 타격을 주었기 때문에 논란이 되고 있다. 2023년 기준으로 오픈서치의 다운로드 건수는 1억을 넘겼으니, 결국 AWS가 성공한 셈이다. 필자는 AWS 서비스를 사용하고 있었으며, AWS..

Spark 가 지원하는 데이터 소스를 읽고 쓰는 방법에 대한 단원이다. 데이터 소스 읽기 spark.read.format("csv")\ .option("mode", "FAILFAST")\ # 읽기 모드 .option("inferSchema", "true")\ .schema(someSchema)\ .load() 데이터를 읽을 때는 DataFrameReader를 사용하며, 이는 SparkSession의 read 속성으로 접근한다. 포맷, 스키마, 읽기 모드, 옵션 과 같은 값들을 지정해주어야 한다. 읽기 모드는 스파크가 형식에 맞지 않는 데이터를 만났을 때의 동작방식을 지정하는 옵션이다. 읽기 모드 permissive : 오류 레코드의 모든 필드를 null로 설정하고 모든 오류 레코드를 _corrupt_re..

구조적 API 의 종류 Dataset DataFrame SQL Table 과 View DataFrame 과 Dataset 의 특징 공통점 : row 와 column 을 가지는 분산 테이블 형태의 컬렉션 Dataset은 결과를 생성하기 위해 어떤 데이터에 어떤 연산을 적용해야 하는지 정의하는 지연 연산의 실행 계획이며 불변성을 가진다. 스키마에 명시된 데이터 타입의 일치 여부를 컴파일 타임에 확인한다. Dataset 은 JVM 기반 언어인 scala 와 java에서만 지원하며, Dataset의 데이터 타입을 정의하려면 scala case class 나 JavaBean 을 사용해야 한다. DataFrame은 액션을 호출하면 스파크가 트랜스포매이션을 실제로 실행하고 반환함. 스키마에 명시된 데이터 타입의 일치..

Dataset - Java, Scala 언어에서 지원하며 정적 데이터 타입에 맞는 코드를 지원하기 위한 Spark 의 구조적 API이다. - Dataset API는 타입 안정성을 지원하므로 초기화에 사용한 class 대신 다른 class로 접근할 수 없다. ex. Dataset[Person] 구조적 스트리밍 - 스트림 처리용 고수준 API - 구조적 스트리밍을 사용하면 구조적 API 로 개발된 배치 모드의 연산을 스트리밍 방식으로 실행할 수 있음. - 데이터를 읽을 때 read 대신 readStream method 를 사용 - maxFiledsPerTrigger 옵션을 추가로 지정 우선 staticDataFrame 을 기존의 csv 파일들을 읽어서 생성한다. 구조적 스트리밍으로 실행하는 예시는 다음과 같..

스파크 기본 아키텍처 스파크는 클러스터의 데이터 처리 작업을 관리하고 조율한다. 스파크가 연산에 사용할 클러스터는 Spark Standalone Cluster Manager, Hadoop YARN, Mesos 와 같은 클러스터 매니저가 관리한다. 유저가 스파크 어플리케이션을 스파크 클러스터 매니저에 submit 하면 클러스터 매니저는 자원을 할당한다. 스파크 어플리케이션 스파크 어플리케이션은 드라이버 프로세스(Driver Process)와 다수의 익스큐터 프로세스 (Executor Process) 로 구성된다. 드라이버 프로세스는 main() 함수를 실행한다. 익스큐터 프로세스는 드라이버 프로세스가 할당한 작업을 수행한다. 그리고 진행 상황을 다시 드라이버 노드에 보고한다. 스파크 API 스파크 API ..

아파치 스파크 (Apache Spark) 란 통합 컴퓨팅 엔진이며 클러스터 환경에서 데이터를 병렬로 처리하는 '라이브러리'들의 집합이다. 1. 컴퓨팅 엔진 여기서 스파크가 '컴퓨팅 엔진' 이라는 의미는 스파크는 저장소 시스템의 데이터를 연산하는 역할만 수행하며 스파크가 직접 빅데이터들을 영구적으로 저장하는 저장소의 역할은 하지 않는다는 의미로 이해할 수 있다. 이러한 점은 Apache Hadoop 과 같은 기존의 빅데이터 플랫폼과의 차별점이 된다. Hadoop의 경우 하둡 파일 시스템과 컴퓨팅 시스템인 맵리듀스라는 두 가지 시스템이 매우 밀접하게 얽혀있어서 둘 중 하나만 단독으로 사용하기 어렵다. 반면, Spark의 경우에는 연산 작업을 수행하는 노드와 저장소를 별도로 구성할 수 있다는 장점이 있다. 2..

Spark 스터디를 시작하면서 실습환경이 필요했는데 현재 회사에서 데이터브릭스를 사용하니까, Spark 개인 공부도 동일한 환경에서 공부를 하면 좋을 것 같아서 데이터브릭스로 결정했다. 그런데 클라우드 서비스인 만큼 과금을 주의해야 하므로 무료 버전 Databricks Community Edition 을 찾아보게 되었다. 시작 방법은 간단한데 데이터브릭스가 상당히 교묘하게 숨겨놔서 주의해야 한다 ! 나와 같은 사람들이 있을 것 같아서 공유한다. https://www.databricks.com/try-databricks Try Databricks - Unified Data Analytics Platform for Data Engineering Discover why businesses are turning..

Apache Airflow 환경 구성 MacOS 환경 Python 3.8.10 버전 단일 local server 위의 환경에서 Apache Airflow 를 설치한다. 우선, airflow 를 설치할 가상 환경을 생성한다. python3 -m venv ./airflow 가상환경을 activate 시켜줍니다. source airflow/bin/activate 다음으로 AIRFLOW_HOME 이라는 환경변수에 airflow 의 경로를 저장합니다. export AIRFLOW_HOME=~/airflow echo 명령어로 환경변수가 잘 설정되었는지 확인합니다. airflow version 에 대한 환경 변수를 설정한다. airflow 2.7 version 은 Python 3.8, 3.9, 3.10, 3.11 을 ..