<jupyter_start><jupyter_text>1교시 스파크 기본 명령어> 스파크의 기본 명령어와 구조에 대해 이해합니다 목차* [1. 스파크를 통한 CSV 파일 읽기](1.-스파크를-통한-CSV-파일-읽기)* [2. 스파크의 2가지 프로그래밍 방식 비교](2.-스파크의-2가지-프로그래밍-방식-비교)* [3. 스파크를 통한 JSON 파일 읽기](3.-스파크를-통한-JSON-파일-읽기)* [4. 뷰 테이블 생성 및 조회](4.-뷰-테이블-생성-및-조회)* [5. 스파크 애플리케이션의 개념 이해](5.-스파크-애플리케이션의-개념-이해)* [6. 스파크 UI](6.-스파크-UI)* [7. M&M 초콜렛 분류 예제](7.-M&M-초콜렛-분류-예제)* [8. 실습문제](8.-실습문제)* [참고자료](참고자료)  1. 스파크를 통한 CSV 파일 읽기> Spark 3.0.1 버전을 기준으로 작성되었습니다. 스파크는 2.0 버전으로 업데이트 되면서 DataFrames 은 Datasets 으로 통합되었고, 기존의 RDD 에서 사용하던 연산 및 기능과 DataFrame 에서 사용하던 것 모두 사용할 수 있습니다. 스파크 데이터 모델은 RDD (Spark1.0) —> Dataframe(Spark1.3) —> Dataset(Spark1.6) 형태로 업그레이드 되었으나, 본문에서 일부 DataFrames 와 DataSets 가 거의 유사하여, 일부 혼용되어 사용되는 경우가 있을 수 있습니다.<jupyter_code>from pyspark.sql import *
from pyspark.sql.functions import *
from pyspark.sql.types import *
from IPython.display import display, display_pretty, clear_output, JSON

spark = (
    SparkSession
    .builder
    .config("spark.sql.session.timeZone", "Asia/Seoul")
    .getOrCreate()
)
# 노트북에서 테이블 형태로 데이터 프레임 출력을 위한 설정을 합니다
spark.conf.set("spark.sql.repl.eagerEval.enabled", True) # display enabled
spark.conf.set("spark.sql.repl.eagerEval.truncate", 100) # display output columns size
# !which python
!/opt/conda/bin/python --version

print("spark.version: {}".format((spark.version)))
spark<jupyter_output>Python 3.8.6
spark.version: 3.0.1<jupyter_text>스파크 사용 관련 팁 여러 줄의 코드 작성* python 코드의 경우 괄호로 (python) 묶으면 이스케이핑(\) 하지 않아도 됩니다* sql 문의 경우  """sql""" 으로 묶으면 이스케이핑(\)하지 않아도 됩니다 데이터 출력 함수* DataFrame.show() - 기본 제공 함수이며, show(n=limit) 통하여 최대 출력을 직접 조정할 수 있으나, 한글 출력 시에 표가 깨지는 경우가 있습니다* display(DataFrame) - Ipython 함수이며, limit 출력을 위해서는 limit 를 걸어야 하지만, 한글 출력에도 깨지지 않습니다<jupyter_code>## 파이썬 코드 여러 줄 작성
json = (
    spark
    .read
    .json("data/tmp/simple.json")
    .limit(2)
)

## 스파크 SQL 여러 줄 작성
json.createOrReplaceTempView("simple")
spark.sql("""
    select * 
    from simple
""")

json.printSchema()
emp_id = json.select("emp_id")

## 표준 데이터 출력함수
json.show()
emp_id.show()

## 노트북 출력함수 
display(json)
display(emp_id)<jupyter_output>root
 |-- emp_id: long (nullable = true)
 |-- emp_name: string (nullable = true)

+------+--------+
|emp_id|emp_name|
+------+--------+
|     1|엘지전자|
|     2|엘지화학|
+------+--------+

+------+
|emp_id|
+------+
|     1|
|     2|
+------+<jupyter_text>컨테이너 기반 노트북> 컨테이너 내부에 존재하는 파일 등에 대해 직접 접근이 가능합니다<jupyter_code>strings = spark.read.text("../requirements.txt")
strings.show(5, truncate=False)
count = strings.count()
print("count of word is {}".format(count))

strings.printSchema()
from pyspark.sql.dataframe import DataFrame
from pyspark.sql.column import Column

assert(type(strings) == DataFrame)
assert(type(strings.value) == Column) # 현재 strings 데이터프레임의 스키마에 value 라는 하나의 컬럼만 존재합니다
# help(strings) # 데이터프레임은 Structured API 를 통해 Row 타입의 레코드를 다루는 함수를 이용하고
# help(strings.value) # 컬럼은 컬럼과의 비교 혹은 포함된 문자열을 다루는 contains 같은 함수를 사용합니다<jupyter_output><empty_output><jupyter_text>아무런 옵션을 주지 않는 경우 스파크가 알아서 컬럼 이름과 데이터 타입을 (string) 지정합니다<jupyter_code>log_access = spark.read.csv("data/log_access.csv")
log_access.printSchema()
log_access.show()<jupyter_output>root
 |-- _c0: string (nullable = true)
 |-- _c1: string (nullable = true)
 |-- _c2: string (nullable = true)

+----------+-----+------+
|       _c0|  _c1|   _c2|
+----------+-----+------+
|    a_time|a_uid|  a_id|
|1603645200|    1| login|
|1603647200|    1|logout|
|1603649200|    2| login|
|1603650200|    2|logout|
|1603653200|    2| login|
|1603657200|    3| login|
|1603659200|    3|logout|
|1603660200|    4| login|
|1603664200|    4|logout|
|1603664500|    4| login|
|1603666500|    5| login|
|1603669500|    5|logout|
|1603670500|    6| login|
|1603673500|    7| login|
|1603674500|    8| login|
|1603675500|    9| login|
+----------+-----+------+<jupyter_text>첫 번째 라인에 헤더가 포함되어 있는 경우 아래와 같이 header option 을 지정하면 컬럼 명을 가져올 수 있습니다<jupyter_code>log_access = spark.read.option("header", "true").csv("data/log_access.csv")
log_access.printSchema()
log_access.show()<jupyter_output>root
 |-- a_time: string (nullable = true)
 |-- a_uid: string (nullable = true)
 |-- a_id: string (nullable = true)

+----------+-----+------+
|    a_time|a_uid|  a_id|
+----------+-----+------+
|1603645200|    1| login|
|1603647200|    1|logout|
|1603649200|    2| login|
|1603650200|    2|logout|
|1603653200|    2| login|
|1603657200|    3| login|
|1603659200|    3|logout|
|1603660200|    4| login|
|1603664200|    4|logout|
|1603664500|    4| login|
|1603666500|    5| login|
|1603669500|    5|logout|
|1603670500|    6| login|
|1603673500|    7| login|
|1603674500|    8| login|
|1603675500|    9| login|
+----------+-----+------+<jupyter_text>inferSchema 옵션으로 데이터 값을 확인하고 스파크가 데이터 타입을 추정하게 할 수 있습니다<jupyter_code>log_access = spark.read.option("header", "true").option("inferSchema", "true").csv("data/log_access.csv")
log_access.printSchema()
log_access.show()<jupyter_output>root
 |-- a_time: integer (nullable = true)
 |-- a_uid: integer (nullable = true)
 |-- a_id: string (nullable = true)

+----------+-----+------+
|    a_time|a_uid|  a_id|
+----------+-----+------+
|1603645200|    1| login|
|1603647200|    1|logout|
|1603649200|    2| login|
|1603650200|    2|logout|
|1603653200|    2| login|
|1603657200|    3| login|
|1603659200|    3|logout|
|1603660200|    4| login|
|1603664200|    4|logout|
|1603664500|    4| login|
|1603666500|    5| login|
|1603669500|    5|logout|
|1603670500|    6| login|
|1603673500|    7| login|
|1603674500|    8| login|
|1603675500|    9| login|
+----------+-----+------+<jupyter_text>1. [기본] "data/flight-data/csv/2010-summary.csv" 파일의 스키마와 데이터 10건을 출력하세요[정답] 출력 결과 확인 > 아래와 유사하게 방식으로 작성 되었다면 정답입니다```pythondf1 = (    spark    .read    .option("header", "true")    .option("inferSchema", "true")    .csv("data/flight-data/csv/2010-summary.csv"))    df1.printSchema()df1.show(10)```<jupyter_code># 여기에 실습 코드를 작성하고 실행하세요 (Shift+Enter)
df1 = (
    spark
    .read
    .option("header", "true")
    .option("inferSchema", "true")
    .csv("data/flight-data/csv/2010-summary.csv")
)

df1.printSchema()
df1.show(10)<jupyter_output>root
 |-- DEST_COUNTRY_NAME: string (nullable = true)
 |-- ORIGIN_COUNTRY_NAME: string (nullable = true)
 |-- count: integer (nullable = true)

+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|    United States|            Romania|    1|
|    United States|            Ireland|  264|
|    United States|              India|   69|
|            Egypt|      United States|   24|
|Equatorial Guinea|      United States|    1|
|    United States|          Singapore|   25|
|    United States|            Grenada|   54|
|       Costa Rica|      United States|  477|
|          Senegal|      United States|   29|
|    United States|   Marshall Islands|   44|
+-----------------+-------------------+-----+
only showing top 10 rows<jupyter_text>2. 스파크의 2가지 프로그래밍 방식 비교  하나. 구조화된 API 호출을 통해 데이터를 출력하는 방법> 출력 시에 bigint 값인 날짜는 아래와 같이 from_unixtime 및 to_timestamp 함수를 통해 변환할 수 있습니다.<jupyter_code>from pyspark.sql.functions import unix_timestamp, from_unixtime, to_timestamp, to_date, col, lit

df = spark.read.option("inferSchema", "true").json("data/activity-data")

# 구조화된 API 를 통한 구문
timestamp = df.select(
    "Arrival_Time",
    to_timestamp(from_unixtime(col('Arrival_Time') / lit(1000)), 'yyyy-MM-dd HH:mm:ss').alias('String_Datetime')
)
timestamp.show(5)<jupyter_output>+-------------+-------------------+
| Arrival_Time|    String_Datetime|
+-------------+-------------------+
|1424686734992|2015-02-23 19:18:54|
|1424686735190|2015-02-23 19:18:55|
|1424686735395|2015-02-23 19:18:55|
|1424686735593|2015-02-23 19:18:55|
|1424686735795|2015-02-23 19:18:55|
+-------------+-------------------+
only showing top 5 rows<jupyter_text>둘. 표현식 형식으로 그대로 사용하여 출력하는 방법> 컬럼(col) 혹은 함수(concat 등)를 직접 사용하는 방식을 **구조화된 API** 를 사용한다고 말하고 SQL 구문으로 표현하는 방식을 **SQL 표현식**을 사용한다고 말합니다<jupyter_code># SQL Expression 통한 구문
ts = df.selectExpr(
    "Arrival_Time",
    "to_timestamp(from_unixtime(Arrival_Time / 1000), 'yyyy-MM-dd HH:mm:ss') as String_Datetime"
)
ts.show(5)<jupyter_output>+-------------+-------------------+
| Arrival_Time|    String_Datetime|
+-------------+-------------------+
|1424686734992|2015-02-23 19:18:54|
|1424686735190|2015-02-23 19:18:55|
|1424686735395|2015-02-23 19:18:55|
|1424686735593|2015-02-23 19:18:55|
|1424686735795|2015-02-23 19:18:55|
+-------------+-------------------+
only showing top 5 rows<jupyter_text>2. [기본] "data/activity-data" 경로에 저장된 json 파일을 읽고 1. 스키마를 출력하세요 2. 10건의 데이터를 출력하세요 3. 'Creation_Time' 컬럼을 년월일 포맷으로 'String_Creation_Date' 컬럼을 출력하세요> 단, Creation_Time 은 Arrival_Time 과 정밀도가 달라서 1000 이 아니라 `1000000000` 을 나누어 주어야 합니다[실습2] 출력 결과 확인 > 아래와 유사하게 방식으로 작성 되었다면 정답입니다```pythondf2 = (    spark    .read    .option("header", "true")    .option("inferSchema", "true")    .json("data/activity-data"))    df2.printSchema()display(df2.limit(3))answer = df2.limit(3).selectExpr(    "Creation_Time",    "to_timestamp(from_unixtime(Creation_Time / 1000000000), 'yyyy-MM-dd HH:mm:ss') as String_Creation_Date")answer.show(10)```<jupyter_code># 여기에 실습 코드를 작성하고 실행하세요 (Shift+Enter)
df2 = (
    spark
    .read
    .option("header", "true")
    .option("inferSchema", "true")
    .json("data/activity-data")
)

df2.printSchema()
display(df2.limit(3))
answer = df2.limit(3).selectExpr(
    "Creation_Time",
    "to_timestamp(from_unixtime(Creation_Time / 1000000000), 'yyyy-MM-dd HH:mm:ss') as String_Creation_Date"
)
answer.show(10)<jupyter_output>root
 |-- Arrival_Time: long (nullable = true)
 |-- Creation_Time: long (nullable = true)
 |-- Device: string (nullable = true)
 |-- Index: long (nullable = true)
 |-- Model: string (nullable = true)
 |-- User: string (nullable = true)
 |-- gt: string (nullable = true)
 |-- x: double (nullable = true)
 |-- y: double (nullable = true)
 |-- z: double (nullable = true)<jupyter_text>Select 뿐만 아니라 filter 의 경우도 Expression 을 사용할 수 있습니다<jupyter_code>df.filter(col("index") > 100).select("index", "user").groupBy("user").count().show()
# 대부분의 구문에서 표현식을 통해 처리할 수 있도록 내부적으로 2가지 방식에 대해 모두 구현되어 있습니다. 
df.filter("index > 100").select("index", "user").groupBy("user").count().show()<jupyter_output>+----+-----+
|user|count|
+----+-----+
|   g|91650|
|   f|92030|
|   e|96000|
|   h|77300|
|   d|81220|
|   c|77130|
|   i|92530|
|   b|91210|
|   a|80824|
+----+-----+

+----+-----+
|user|count|
+----+-----+
|   g|91650|
|   f|92030|
|   e|96000|
|   h|77300|
|   d|81220|
|   c|77130|
|   i|92530|
|   b|91210|
|   a|80824|
+----+-----+<jupyter_text>3. 스파크를 통한 JSON 파일 읽기> 추후에 학습하게 될 예정인 filter 및 groupBy 구문이 사용되고 있는데요, 조건을 통해 데이터를 줄이고(filter), 특정 컬럼별 집계(groupBy)를 위한 연산자입니다<jupyter_code>json = spark.read.json("data/activity-data")
users = json.filter("index > 100").select("index", "user").groupBy("user").count()
users.show(5)<jupyter_output>+----+-----+
|user|count|
+----+-----+
|   g|91650|
|   f|92030|
|   e|96000|
|   h|77300|
|   d|81220|
+----+-----+
only showing top 5 rows<jupyter_text>3. [기본] "data/activity-data" 경로의 JSON 데이터를 읽고 1. 스키마를 출력하세요 2. 10건의 데이터를 출력하세요 3. index 가 10000 미만의 이용자('user')별 빈도수를 구하세요[실습3] 출력 결과 확인 > 아래와 유사하게 방식으로 작성 되었다면 정답입니다```pythondf3 = (    spark    .read    .option("header", "true")    .option("inferSchema", "true")    .json("data/activity-data"))    df3.printSchema()answer = df3.filter("index < 10000").groupBy("user").count()answer.show(10)```<jupyter_code># 여기에 실습 코드를 작성하고 실행하세요 (Shift+Enter)
df3 = (
    spark
    .read
    .option("header", "true")
    .option("inferSchema", "true")
    .json("data/activity-data")
)

df3.printSchema()
answer = df3.filter("index < 10000").groupBy("user").count()
answer.show(10)<jupyter_output>root
 |-- Arrival_Time: long (nullable = true)
 |-- Creation_Time: long (nullable = true)
 |-- Device: string (nullable = true)
 |-- Index: long (nullable = true)
 |-- Model: string (nullable = true)
 |-- User: string (nullable = true)
 |-- gt: string (nullable = true)
 |-- x: double (nullable = true)
 |-- y: double (nullable = true)
 |-- z: double (nullable = true)

+----+-----+
|user|count|
+----+-----+
|   g| 2506|
|   f| 2490|
|   e| 2501|
|   h| 2500|
|   d| 2499|
|   c| 2494|
|   i| 2500|
|   b| 2500|
|   a| 2501|
+----+-----+<jupyter_text>4. 뷰 테이블 생성 및 조회> 이미 생성된 데이터 프레임을 통해서 현재 세션에서만 조회 가능한 임시 뷰 테이블을 만들어 SQL 질의가 가능합니다.<jupyter_code>users.createOrReplaceTempView("users")
spark.sql("select * from users where count is not null and count > 9000 order by count desc").show(5)<jupyter_output>+----+-----+
|user|count|
+----+-----+
|   e|96000|
|   i|92530|
|   f|92030|
|   g|91650|
|   b|91210|
+----+-----+
only showing top 5 rows<jupyter_text>4. [기본] "data/flight-data/json/2015-summary.json" 경로의 JSON 데이터를 읽고 1. `2015_summary` 라는 임시 테이블을 생성하세요 2. spark sql 구문을 이용하여 10 건의 데이터를 출력하세요[실습4] 출력 결과 확인 > 아래와 유사하게 방식으로 작성 되었다면 정답입니다```pythondf4 = (    spark    .read    .option("header", "true")    .option("inferSchema", "true")    .json("data/flight-data/json/2015-summary.json"))    df4.printSchema()answer = df4.createOrReplaceTempView("2015_summary")spark.sql("select * from 2015_summary").show(10)```<jupyter_code># 여기에 실습 코드를 작성하고 실행하세요 (Shift+Enter)
df4 = (
    spark
    .read
    .option("header", "true")
    .option("inferSchema", "true")
    .json("data/flight-data/json/2015-summary.json")
)

df4.printSchema()
answer = df4.createOrReplaceTempView("2015_summary")
spark.sql("select * from 2015_summary").show(10)<jupyter_output>root
 |-- DEST_COUNTRY_NAME: string (nullable = true)
 |-- ORIGIN_COUNTRY_NAME: string (nullable = true)
 |-- count: long (nullable = true)

+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|    United States|            Romania|   15|
|    United States|            Croatia|    1|
|    United States|            Ireland|  344|
|            Egypt|      United States|   15|
|    United States|              India|   62|
|    United States|          Singapore|    1|
|    United States|            Grenada|   62|
|       Costa Rica|      United States|  588|
|          Senegal|      United States|   40|
|          Moldova|      United States|    1|
+-----------------+-------------------+-----+
only showing top 10 rows<jupyter_text>JSON 파일을 읽는 여러가지 방법<jupyter_code># 스키마 확인 - 3가지 모두 동일한 결과를 얻을 수 있으며 편한 방식을 선택하시면 됩니다
df = spark.read.format("json").load("./data/flight-data/json/2015-summary.json") # 미국 교통통계국이 제공하는 항공운항 데이터
df.printSchema()

df2 = spark.read.load("./data/flight-data/json/2015-summary.json", format="json")
df2.printSchema()

df3 = spark.read.json("./data/flight-data/json/2015-summary.json")
df3.printSchema()<jupyter_output>root
 |-- DEST_COUNTRY_NAME: string (nullable = true)
 |-- ORIGIN_COUNTRY_NAME: string (nullable = true)
 |-- count: long (nullable = true)

root
 |-- DEST_COUNTRY_NAME: string (nullable = true)
 |-- ORIGIN_COUNTRY_NAME: string (nullable = true)
 |-- count: long (nullable = true)

root
 |-- DEST_COUNTRY_NAME: string (nullable = true)
 |-- ORIGIN_COUNTRY_NAME: string (nullable = true)
 |-- count: long (nullable = true)<jupyter_text>5. 스파크 애플리케이션의 개념 이해  스파크에서 반드시 알아야 할 객체와 개념| 구분 | 설명 | 기타 ||---|---|---|| Application | 스파크 프레임워크를 통해 빌드한 프로그램. 전체 작업을 관리하는 Driver 와 Executors 상에서 수행되는 프로그램으로 구분합니다 | - || SparkSession | 스파크의 모든 기능을 사용하기 위해 생성하는 객체 | - || Job | 하나의 액션(save, collect 등)을 수행하기 위해 여러개의 타스크로 구성된 병렬처리 단위 | DAG 혹은 Spark Execution Plan || Stage | 하나의 잡은 다수의 스테이지라는 것으로 구성되며, 하나의 스테이지는 다수의 타스크 들로 구성됩니다 | - || Task | 스파크 익스큐터에 보내지는 하나의 작업 단위 | 하나의 Core 혹은 Partition 단위의 작업 | 스파크의 변환(Transformation)과 액션(Action)| 구분 | 설명 | 기타 ||---|---|---|| Transformation | 원본 데이터의 변경 없이 새로운 데이터프레임을 생성하는 모든 작업을 말하며 모든 변환 작업들은 lazily evaluated 되며 lineage 를 유지합니다 | select, filter, join, groupBy, orderBy || Action | 여태까지 지연된 변환 작업을 트리거링하는 동작을 말하며, 데이터의 조회 및 저장 등의 작업을 말합니다 | show, take, count, collect, save |> Lineage : 연속된 변환 파이프라인이 액션을 만나기 전까지의 모든 이력 혹은 히스토리 정보를 모두 저장하고 있는 객체를 말하며, 스파크는 이렇게 체인을 구성한 변환 작업을 통해 **쿼리 최적화**를 수행할 수 있으며, 데이터 불변성(Immutability)를 통해서 **내결함성(Fault Tolerance)**을 가질 수 있습니다. 좁은 변환과 넓은 변환 - Narrow and Wide Transformation> 위에서 언급한 최적화(Optimization) 과정은 여러 오퍼레이션들을 다시 여러 스테이지로 쪼개고, 이러한 스테이지 들이 셔플링이 필요한지, 클러스터간의 데이터 교환이 필요한지 등을 결정하는 문제이며, 변환 작업은 크게 하나의 파티션 내에 수행이 가능한 **좁은 의존성(narrow dependencies)** 과 셔플링이 발생하여 클러스터 전체에 데이터 교환이 필요한 **넓은 의존성(wide dependencies)** 두 가지로 구분합니다![Transformation](images/transformation.png)  6. 스파크 UI> Default 포트는 4040 이므로 http://localhost:4040 에 접속하여 앞에서 배웠던 Narrow, Wide Transformation DAG를 확인합니다<jupyter_code># Narrow Transformation
strings = spark.read.text("../requirements.txt")
jupyter = strings.filter(strings.value.contains("jupyter"))
jupyter.show(truncate=False)
# Wide Transformation
user = spark.read.option("header", "true").option("inferSchema", "true").csv("data/tbl_user.csv")
count = user.groupBy("u_gender").count()
count.show(truncate=False)<jupyter_output>+--------+-----+
|u_gender|count|
+--------+-----+
|여      |3    |
|남      |6    |
+--------+-----+<jupyter_text>| Narrow | Wide ||---|---||![narrow](images/narrow.png)|![wide](images/wide.png)|  7. M&M 초콜렛 분류 예제 (참고용)> [Learning Spark 2nd Edition](https://github.com/psyoblade/LearningSparkV2?organization=psyoblade&organization=psyoblade) 에서 제공하는 data bricks dataset 예제 가운데 미국의 주 별 M&M 초콜렛 판매량을 집계하는 예제를 작성합니다<jupyter_code>mnm_df = spark.read.option("header", "true").option("inferSchema", "true").csv("data/databricks/mnm_dataset.csv")
mnm_df.printSchema()
mnm_df.show(truncate=False)
from pyspark.sql.functions import *

# We use the DataFrame high-level APIs. Note
# that we don't use RDDs at all. Because some of Spark's
# functions return the same object, we can chain function calls.
# 1. Select from the DataFrame the fields "State", "Color", and "Count"
# 2. Since we want to group each state and its M&M color count,
# we use groupBy()
# 3. Aggregate counts of all colors and groupBy() State and Color
# 4 orderBy() in descending order
count_mnm_df = (mnm_df.select("State", "Color", "Count") \
.groupBy("State", "Color") \
.agg(count("Count").alias("Total")) \
.orderBy("Total", ascending=False))
# Show the resulting aggregations for all the states and colors;
# a total count of each color per state.
# Note show() is an action, which will trigger the above
# query to be executed.
count_mnm_df.show(n=60, truncate=False)
print("Total Rows = %d" % (count_mnm_df.count()))

# While the above code aggregated and counted for all
# the states, what if we just want to see the data for
# a single state, e.g., CA?
# 1. Select from all rows in the DataFrame
# 2. Filter only CA state
# 3. groupBy() State and Color as we did above
# 4. Aggregate the counts for each color
# 5. orderBy() in descending order
# Find the aggregate count for California by filtering
ca_count_mnm_df = (mnm_df.select("State", "Color", "Count") \
.where(mnm_df.State == "CA") \
.groupBy("State", "Color") \
.agg(count("Count").alias("Total")) \
.orderBy("Total", ascending=False))
# Show the resulting aggregation for California.
# As above, show() is an action that will trigger the execution of the
# entire computation.
ca_count_mnm_df.show(n=10, truncate=False)
# Stop the SparkSession
# spark.stop()<jupyter_output>+-----+------+-----+
|State|Color |Total|
+-----+------+-----+
|CA   |Yellow|1807 |
|WA   |Green |1779 |
|OR   |Orange|1743 |
|TX   |Green |1737 |
|TX   |Red   |1725 |
|CA   |Green |1723 |
|CO   |Yellow|1721 |
|CA   |Brown |1718 |
|CO   |Green |1713 |
|NV   |Orange|1712 |
|TX   |Yellow|1703 |
|NV   |Green |1698 |
|AZ   |Brown |1698 |
|CO   |Blue  |1695 |
|WY   |Green |1695 |
|NM   |Red   |1690 |
|AZ   |Orange|1689 |
|NM   |Yellow|1688 |
|NM   |Brown |1687 |
|UT   |Orange|1684 |
|NM   |Green |1682 |
|UT   |Red   |1680 |
|AZ   |Green |1676 |
|NV   |Yellow|1675 |
|NV   |Blue  |1673 |
|WA   |Red   |1671 |
|WY   |Red   |1670 |
|WA   |Brown |1669 |
|NM   |Orange|1665 |
|WY   |Blue  |1664 |
|WA   |Yellow|1663 |
|WA   |Orange|1658 |
|CA   |Orange|1657 |
|NV   |Brown |1657 |
|CO   |Brown |1656 |
|CA   |Red   |1656 |
|UT   |Blue  |1655 |
|AZ   |Yellow|1654 |
|TX   |Orange|1652 |
|AZ   |Red   |1648 |
|OR   |Blue  |1646 |
|UT   |Yellow|1645 |
|OR   |Red   |1645 |
|CO   |Orange|1642 |
|TX   |Brown [...]<jupyter_text>5. [기본] "data/tbl_user.csv" 경로의 CSV 데이터를 읽고 1. 스키마를 출력하세요 2. `user` 라는 임시 테이블을 생성하세요 3. spark sql 구문을 이용하여 10 건의 데이터를 출력하세요[실습5] 출력 결과 확인 > 아래와 유사하게 방식으로 작성 되었다면 정답입니다```pythondf5 = (    spark    .read    .option("header", "true")    .option("inferSchema", "true")    .csv("data/tbl_user.csv"))    df5.printSchema()answer = df5.createOrReplaceTempView("user")spark.sql("select * from user").show(10)```<jupyter_code># 여기에 실습 코드를 작성하고 실행하세요 (Shift+Enter)
df5 = (
    spark
    .read
    .option("header", "true")
    .option("inferSchema", "true")
    .csv("data/tbl_user.csv")
)

df5.printSchema()
answer = df5.createOrReplaceTempView("user")
spark.sql("select * from user").show(10)<jupyter_output>root
 |-- u_id: integer (nullable = true)
 |-- u_name: string (nullable = true)
 |-- u_gender: string (nullable = true)
 |-- u_signup: integer (nullable = true)

+----+----------+--------+--------+
|u_id|    u_name|u_gender|u_signup|
+----+----------+--------+--------+
|   1|    정휘센|      남|19700808|
|   2|  김싸이언|      남|19710201|
|   3|    박트롬|      여|19951030|
|   4|    청소기|      남|19770329|
|   5|유코드제로|      여|20021029|
|   6|  윤디오스|      남|20040101|
|   7|  임모바일|      남|20040807|
|   8|  조노트북|      여|20161201|
|   9|  최컴퓨터|      남|20201124|
+----+----------+--------+--------+<jupyter_text>6. [기본] "data/tbl_purchase.csv" 경로의 CSV 데이터를 읽고 1. 스키마를 출력하세요 2. `purchase` 라는 임시 테이블을 생성하세요 3. selectExpr 구문 혹은 spark sql 구문을 이용하여 `p_time` 필드를 날짜 함수를 이용하여 식별 가능하도록 데이터를 출력하세요[실습6] 출력 결과 확인 > 아래와 유사하게 방식으로 작성 되었다면 정답입니다```pythondf6 = (    spark    .read    .option("header", "true")    .option("inferSchema", "true")    .csv("data/tbl_purchase.csv"))    df6.printSchema()answer = df6.createOrReplaceTempView("purchase")spark.sql("select from_unixtime(p_time) as p_time from purchase").show(10)```<jupyter_code># 여기에 실습 코드를 작성하고 실행하세요 (Shift+Enter)
df6 = (
    spark
    .read
    .option("header", "true")
    .option("inferSchema", "true")
    .csv("data/tbl_purchase.csv")
)

df6.printSchema()
answer = df6.createOrReplaceTempView("purchase")
spark.sql("select from_unixtime(p_time) as p_time from purchase").show(10)<jupyter_output>root
 |-- p_time: integer (nullable = true)
 |-- p_uid: integer (nullable = true)
 |-- p_id: integer (nullable = true)
 |-- p_name: string (nullable = true)
 |-- p_amount: integer (nullable = true)

+-------------------+
|             p_time|
+-------------------+
|2020-10-26 03:45:50|
|2020-10-26 03:45:50|
|2020-10-26 15:45:55|
|2020-10-26 09:51:40|
|2020-10-26 03:55:55|
|2020-10-26 10:08:20|
|2020-10-26 07:45:55|
|2020-10-26 07:49:15|
+-------------------+
{'fraction_non_alphanumeric': 0.1952564326973569, 'fraction_numerical': 0.07487309644670051, 'mean_word_length': 3.08673104434907, 'pattern_counts': {'":': 0, '<': 64, '<?xml version=': 0, '>': 86, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 3, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '20', 'dir': 'jupyter-structured-clean-dedup', 'id': '118324', 'license': 'Apache-2.0', 'n_tokens_mistral': 11904, 'n_tokens_neox': 12493, 'n_words': 2238, 'path': 'day3/notebooks/lgde-spark-core-1-basic-answer.ipynb', 'repo_name': 'psyoblade/data-engineer-basic-training', 'size': '22852'}