스파크 select, selctExpr, lit, distinct, limit (스파크 정리 5)

 

전편

☞ 2022.09.24 - [Data Engineering/스파크] - 스파크 기본 명령어, 스파크 데이터프레임 명령어 (스파크 정리 4)

 

 

아마도 스파크에서 가장 많이 쓰일 명령어 select. 타겟 데이터프레임에서 보고 싶은 컬럼을 선택할 때 사용된다. 기본적인 기능이나 사용법은 SQL의 SELECT와 유사하다. 이번 포스팅도 역시 스칼라와 파이썬을 사용한다. 모듈이나 패키지를 가져오는 부분 빼고는 스칼라와 파이썬의 코드는 동일하다. 

 

실습을 위한 데이터는 스파크 정리 2 포스팅(2022.09.16 - [Data Engineering/스파크] - 스파크란? 스파크 기본 개념 (스파크 정리 2))을 참고하면 다운받을 수 있다. 일단 데이터 아래 코드처럼 데이터를 읽어서 데이터프레임으로 저장하자.

// in scalar
val df = spark.read.format("json")
  .load("/data/flight-data/json/2015-summary.json")

 

# in python
df = spark.read.format("json").load("/data/flight-data/json/2015-summary.json")

 

 

1. select()

select("컬럼명1", "컬럼명2", ...)

// in Scala
df.select("DEST_COUNTRY_NAME").show(2)
=======================================================================================
>> 출력
+-----------------+
|DEST_COUNTRY_NAME|
+-----------------+
|    United States|
|    United States|
+-----------------+

 

# in Python
df.select("DEST_COUNTRY_NAME", "ORIGIN_COUNTRY_NAME").show(2)
=======================================================================================
>> 출력
+-----------------+-------------------+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|
+-----------------+-------------------+
|    United States|            Romania|
|    United States|            Croatia|
+-----------------+-------------------+

 

select() 함수 안에 컬럼명을 스트링으로 넘겨 줄 수도 있지만 이전 포스팅에서 배운 col(), column() 함수, 또는 expr() 함수를 이용하여 넘겨줄 수도 있다. 

// in Scala
import org.apache.spark.sql.functions.{expr, col, column}
df.select(
    df.col("DEST_COUNTRY_NAME"),
    col("DEST_COUNTRY_NAME"),
    column("DEST_COUNTRY_NAME"),
    'DEST_COUNTRY_NAME,
    $"DEST_COUNTRY_NAME",
    expr("DEST_COUNTRY_NAME"))
  .show(2)

 

# in Python
from pyspark.sql.functions import expr, col, column
df.select(
    expr("DEST_COUNTRY_NAME"),
    col("DEST_COUNTRY_NAME"),
    column("DEST_COUNTRY_NAME"))\
  .show(2)

 

다만 스트링으로 넘겨주는 방식과 col() 등의 함수를 이용해 넘겨주는 방식을 같이 쓰면 안된다. 

// Error!
df.select(col("DEST_COUNTRY_NAME"), "DEST_COUNTRY_NAME")

 

또한 SQL처럼 "AS"를 통해 출력되는 컬럼이름을 바꿔줄 수 있다. 이때는 expr() 함수를 사용해 구문을 통째로 넘겨주어야 한다. 

// in Scala
df.select(expr("DEST_COUNTRY_NAME AS DEST")).show(1)
=======================================================================================
>> 출력
+-------------+
|         DEST|
+-------------+
|United States|
+-------------+

 

 

2. selectExpr()

select() 함수 안에 expr() 함수를 사용하는 경우가 많다보니 스파크에서는 두 개를 합친 함수, selectExpr()을 만들었다.

df.selectExpr("DEST_COUNTRY_NAME as newColumnName", "DEST_COUNTRY_NAME").show()

 

스파크에서도 SQL과 마찬가지로 모든 컬럼을 뜻하는 와일드카드인 "*"를 쓸 수 있다.

// in Scala
df.selectExpr(
    "*", // include all original columns
    "(DEST_COUNTRY_NAME = ORIGIN_COUNTRY_NAME) as withinCountry")
  .show(2)
=======================================================================================
>> 출력
+-----------------+-------------------+-----+-------------+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|withinCountry|
+-----------------+-------------------+-----+-------------+
|    United States|            Romania|   15|        false|
|    United States|            Croatia|    1|        false|
+-----------------+-------------------+-----+-------------+

 

또한 avg(), count() 등의 어그리게이션 함수도 사용 가능하다. 어그리게이션 함수는 추후 좀 더 자세히 다룬다. 

df.selectExpr("avg(count)", "count(distinct(DEST_COUNTRY_NAME))").show(2)

 

 

3. literals: lit()
현재 사용하는 언어의 리터럴 값을 스파크 타입으로 바꾸고 싶을 때는 lit() 함수를 사용하면 된다. 

// in Scala
import org.apache.spark.sql.functions.lit
df.select(expr("*"), lit(1).as("One")).show(2)

 

# in Python
from pyspark.sql.functions import lit
df.select(expr("*"), lit(1).alias("One")).show(2)
=======================================================================================
>> 출력
+-----------------+-------------------+-----+---------+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|numberOne|
+-----------------+-------------------+-----+---------+
|    United States|            Romania|   15|        1|
|    United States|            Croatia|    1|        1|
+-----------------+-------------------+-----+---------+

 

 

4. distinct()

특정 컬럼에 대해서 중복된 값을 빼고 보고 싶다면 distinct를 사용하면 된다. 

df.select("ORIGIN_COUNTRY_NAME", "DEST_COUNTRY_NAME").distinct().count()

 

 

5. limit()

선택한 컬럼에 대해 모든 리코드가 아니라 앞에 n개의 리코드만 보고 싶다면 limit(n)을 사용하면 된다. 

df.limit(5).show()

 

 

다음편

☞ 2022.09.27 - [Data Engineering/스파크] - 스파크 filter, where, sort, sample, split, union (스파크 정리 6)

스파크 filter, where, sort, sample, split, union (스파크 정리 6)