[Spark] SQL and DataFrame

스파크 SQL

소개

각 레코드의 필드 구성에 대해 모두 알고 있을 경우 스파크 SQL 을 더 쉽고 효과적으로 불러오고 쿼리를 쓸 수 있다.

스파크 SQL 주요 기능 3가지

1. DataFrame 추상화 클래스 제공

2. 다양한 구조적 포맷의 데이터를 읽고 쓸 수 있다 (JSON, Hive, 파케이(parquet)

3. 내부 JDBC/ODBC 나 외부 Tableau (태블루) 같은 BI 툴 등을 써서 스파크 SQL 을 통해 SQL 로 데이터를 질의 할 수 있다.

SchemaRDD

Row 객체의 RDD 

각 아이템은 Record 를 의미

RDD 와 유사해 보이지만 내부적으로는 좀더 효율적인 방법으로 저장하고 Scheme 이점 활용한다

RDD 에서 제공되지 않는 Operation 도 제공

DataFrame은 이전버젼(1.0~1.2)에서 쓰이는 SchemaRDD가 진화한 것

SchemaRDD 로 코딩되어 있어도 1.3이상에서는 DataFrame으로 쓰는것과 같다.

Dataset 은 SQL 의 최적화된 실행 엔진의 이점을 이용하여 RDD (강력한 타이핑, 람다기능)의 이점을 제공하는 Spark 1.6 에서 추가된 새로운 인터페이스.

그외

1. 스파크 SQL 은 하둡의SQL 엔진인 apache Hive 를 포함하거나 또는 포함하지 않고서 빌드가 가능하다.

2. 하이브 의존성 충돌은 주의 해야한다. spark-hive_2.10, spark-sql_2.10

-> SQLContext 는 스파크 SQL의 기능을 제공하지만 하이브에 의존하지는 않는다.

3. HiveQL 은 스파크 SQL 을 쓰기위해 추천하는 질의 언어

구조

그림1

아래는 스파크의 전체 구조입니다.

우리는 본 문서에서 SQL 에 대해서 살펴볼 예정입니다. 

아래 그림2에서 Spark SQL 구조에 대해서 알아봅시다.

그림2

스파크 SQL 의 적용

Hive, JSON, Parquet 등으로 수집된 Spark SQL 을 태블루 등이 JDBC 등을 이용하여 질의 하는 내용이다.

그림3

다음 그림은 스파크SQL의 다른 세부 구조를 보여줍니다.

즉 API, RDD, DATA Source 가 아키텍쳐에 포함되어 있다.

DataFrame

1 큰 클러스터 단일 노드 클러스터에서 페타 바이트에 킬로바이트 크기의 데이터를 처리 할 수 있는 능력.

2 다른 데이터 형식 (아 브로, CSV, 탄성 검색, 카산드라) 및 스토리지 시스템 (HDFS, HIVE 테이블, MySQL은, 등)을 지원합니다.

3 스파크의 SQL 촉매 최적화 (트리 변환 프레임 워크)을 통해 기술 최적화 및 코드 생성의 주.

4 쉽게 스파크 코어를 통해 모든 빅 데이터 툴과 프레임 워크와 통합 할 수 있습니다.

5 파이썬, 자바, 스칼라, 및 R 프로그래밍을위한 API를 제공합니다.

Databricks 의 데이터 프레임 소개

https://databricks.com/blog/2015/02/17/introducing-dataframes-in-spark-for-large-scale-data-science.html

 DataFrame은 JSON 파일, Parquet 파일, 하이브 테이블을 비롯하여 가장 많이 사용되는 형식의 데이터를 읽는 것을 지원합니다. JDBC를 통해 로컬 파일 시스템, 분산 파일 시스템 (HDFS), 클라우드 스토리지 (S3) 및 외부 관계형 데이터베이스 시스템에서 읽을 수 있습니다. 또한 Spark SQL의 외부 데이터 소스 API 를 통해 DataFrames를 확장하여 타사 데이터 형식이나 소스를 지원할 수 있습니다. 기존의 타사 확장 프로그램에는 Avro, CSV, ElasticSearch 및 Cassandra가 이미 포함되어 있습니다.

데이터 소스에 대한 DataFrames의 지원으로 응용 프로그램은 서로 다른 소스의 데이터 (데이터베이스 시스템에서 페더 레이 티드 쿼리 처리라고 함)를 쉽게 결합 할 수 있습니다.예를 들어, 다음 코드 스 니펫은 S3에 저장된 사이트의 텍스트 트래픽 로그를 PostgreSQL 데이터베이스와 조인하여 각 사용자가 사이트를 방문한 횟수를 계산합니다.

users = context.jdbc(“jdbc:postgresql:production”, “users”)

logs = context.load(“/path/to/traffic.log”)

logs.join(users, logs.userId == users.userId, “left_outer”) \

.groupBy(“userId”).agg({“*”: “count”})

DataFrame에 대한 계산이 시작되기 전에 Catalyst Optimizer 는 DataFrame을 구축하는 데 사용 된 작업을 실제 계획으로 컴파일하여 실행합니다. 옵티마이 저는 조작의 의미 및 데이터 구조를 이해하므로 계산 속도를 높이기 위해 지능적인 결정을 내릴 수 있습니다.

예제>

scala> val sqlcontext = new org.apache.spark.sql.SQLContext(sc)

scala> val dfs = sqlContext.read.json("employee.json")

scala> dfs.show()

scala> dfs.select("name").show()

scala> dfs.groupBy("age").count().show()

참조

1 http://spark.apache.org/docs/latest/sql-programming-guide.html

2 Lerning Spark 서적

3 http://www.w3ii.com/ko/spark_sql/spark_sql_quick_guide.html