Spark 란버클리 대학의 AMPLab에서 아파치 오픈소스 프로젝트대규모 데이터를 처리하기 위한 클러스터 컴퓨팅 프레임워크 Java로 개발되었으며 Python, Sql, Scala등의 언어들을 지원하여, 어떤 언어로 개발하든 성능에 큰 이슈 없도록 설계Hadoop와 달리 MapReduce 로직중 Map를 메모리에서 처리하기에 Hadoop보다 속도가 빠름빅데이터 처리 관련 다양한 기능 제공https://spark.apache.org/releases/spark-release-3-5-0.html Spark Release 3.5.0 | Apache SparkSpark Release 3.5.0 Apache Spark 3.5.0 is the sixth release in the 3.x series. With sig..

카탈로그 테이블과 뷰에 관한 메타 데이터 관리 메타 데이터 : 데이터에 관한 구조화된 데이터, 다른 데이터를 설명해 주는 데이터이다. 기본으로 메모리 기반 카탈로그 제공 - 세션이 끝나면 사라짐 Hive와 호환되는 카탈로그 제공 - Persistent 테이블 관리 방식 테이블들은 데이터베이스라 부르는 폴더와 같은 구조로 관리 (2단계) 메모리 기반 테이블/뷰 임시 테이블 스토리지 기반 테이블 기본적으로 HDFS와 Parquet 포맷을 사용 Hive와 호환되는 메타스토어 사용 두 종류의 테이블이 존재 (Hive와 동일한 개념) Managed Table Spark이 실제 데이터와 메타 데이터 모두 관리 Unmanaged (External) Table Spark이 메타 데이터만 관리 Spark SQL 스토리지..

UDF(User Defined Function) DataFrame이나 SQL에서 적용할 수 있는 사용자 정의 함수 Scalar 함수 vs. Aggregation 함수 Scalar 함수 예: UPPER, LOWER, … Aggregation 함수 (UDAF) 예: SUM, MIN, MAX 함수 등록 pyspark.sql.functions.udf DataFrame에서만 사용 가능 spark.udf.register SQL 모두에서 사용 가능 Sparksession 생성 from pyspark.sql import SparkSession spark = SparkSession \ .builder \ .appName("Python Spark UDF") \ .getOrCreate() 데이터 생성 columns = ["..

JOIN SQL 조인은 두 개 혹은 그 이상의 테이블들을 공통 필드를 가지고 Merge 스타 스키마로 구성된 테이블들로 분산되어 있던 정보를 통합하는데 사용 왼쪽 테이블을 LEFT라고 하고 오른쪽 테이블을 RIGHT이라고 하면 JOIN의 결과는 방식에 따라 양쪽의 필드를 모두 가진 새로운 테이블을 생성 조인의 방식에 따라 어떤 레코드들이 선택되는지, 어떤 필드들이 채워지가 달라짐 Join의 종류 SparkSession 생성 및 사용할 테이블 생성 from pyspark.sql import SparkSession from pyspark import SparkConf conf = SparkConf() conf.set("spark.app.name", "sparksql join") conf.set("spark...

Spark SQL Spark SQL은 구조화된 데이터 처리를 위한 Spark 모듈 데이터 프레임 작업을 SQL로 처리 가능 데이터프레임에 테이블 이름 지정 후 sql함수 사용가능 판다스에도 pandasql 모듈의 sqldf 함수를 이용하는 동일한 패턴 존재 HQL(Hive Query Language)과 호환 제공 Hive 테이블들을 읽고 쓸 수 있음 (Hive Metastore) Spark SQL 의 장점 (DataFrame에 비해) SQL로 가능한 작업이라면 DataFrame을 사용할 이유가 없음 물론 두개를 동시에 사용할 수 있음 Familiarity/Readability SQL이 가독성이 더 좋고 더 많은 사람들이 사용가능 Optimization Spark SQL 엔진이 최적화하기 더 좋음 (SQL..

이번 포스팅에서는 아래 데이터를 활용합니다. 여기서 처리할 데이터는 measure_type가 "TMIN"이고 온도("temperature")가 가장 낮은 "stationID"를 추룰하는 작업을 할 것이다. Pyspark 1. 우선 SparkSession과 SparkConf를 설정한다. from pyspark.sql import SparkSession from pyspark import SparkConf conf = SparkConf() conf.set("spark.app.name", "PySpark DataFrame #1") conf.set("spark.master", "local[*]") spark = SparkSession.builder\ .config(conf=conf)\ .getOrCreate()..