はじめに

Spark 、難しいと思われがちですが、基本的なデータエンジニアリングをするだけであれば SQL で完遂できますので、SQL ユーザーに馴染みやすいお作法を紹介します。
（私の周りも SQLによる ETL 処理を開発して DWH を構築してきたエンジニアが多数派です）

データエンジニアリング例

このような3つのテーブルを使ってデータエンジニアリングをする例で紹介します。

導出結果は顧客ごとの注文回数です。

基本のテーブル読み取りと結果の確認

テーブルを読み取ってノートブック上にデータを読みこむには、テーブルのオプションからデータの読み込みを選択します。
SELECT 文を実行するため、spark.sql関数が入力されます。SQLの結果が変数dfに格納されます。変数名は任意ですがデータフレームをdfと表現して接頭辞か接尾辞として利用することが多いです。
display(df)を実行するとデータが表示されます。

このように spark.sql と display が対話型の SQL 実行で重要な構文となります

WITH句を使用する

SparkSQLは WITH 句（CTE）に対応しているので、これを実行します。

SQL を実行したいだけの場合は、%%sql をセルの先頭に付与すると、SQLセルとして扱われます。

sql


%%sql

with orders as (

    select
        id as order_id,
        user_id as customer_id,
        order_date,
        status
    from raw_orders

)

    select
        customer_id,
        min(order_date) as first_order,
        max(order_date) as most_recent_order,
        count(order_id) as number_of_orders
    from orders
    group by customer_id

実行すると　 display の時と同様に結果が表示されます。

SQL での書き込み

テーブルの同時作成（saveAsTable）は　CREATE TABLE AS SELECT (CTAS)、
テーブルへの書き込みは INSERT INTO または INSERT OVERWRITE （全件上書き）や、UPDATE, MERGE INTO にて行います。

CTASの場合、以下のように記述可能です。

sql

%%sql

create table customer_orders using delta  
as 
with orders as (
    select
        id as order_id,
        user_id as customer_id,
        order_date,
        status
    from raw_orders
)
select
    customer_id,
    min(order_date) as first_order,
    max(order_date) as most_recent_order,
    count(order_id) as number_of_orders
from orders
group by customer_id

テーブルが存在する場合は、DML が利用可能です。

sql


%%sql
 
with orders as (
    select
        id as order_id,
        user_id as customer_id,
        order_date,
        status
    from raw_orders
)

insert overwrite customer_orders
select
    customer_id,
    min(order_date) as first_order,
    max(order_date) as most_recent_order,
    count(order_id) as number_of_orders
from orders
group by customer_id

pyspark の処理とSQLでの処理を織り交ぜる

ノートブックの中では二つの言語を行き来することができます。

sqlの処理結果にpython処理を加えたい場合

pyspark

df = spark.sql("""

with orders as (
    select
        id as order_id,
        user_id as customer_id,
        order_date,
        status
    from raw_orders
)

select
    customer_id,
    min(order_date) as first_order,
    max(order_date) as most_recent_order,
    count(order_id) as number_of_orders
from orders
group by customer_id

""")

import pyspark.sql.functions as F 

df_add_column =  df.withColumn("add_column1",F.lit("追加列"))

display(df_add_column)

pyspark の処理結果をSQLで読みたい場合, createOrReplaceTempView を使用すると簡単です。
pyspark
```
df_add_column.createOrReplaceTempView("sql_tempview")
```
sql
```
%%sql

select * from sql_tempview
```

with句の代わりにcreateOrReplaceTempView をチェーンする

createOrReplaceTempViewを CTEのように使うことができます。

pyspark


ch1 = spark.sql("""
    select
        id as order_id,
        user_id as customer_id,
        order_date,
        status
    from raw_orders
""")
ch1.createOrReplaceTempView("temp_orders")

pyspark

ch2 = spark.sql("""
    select
        customer_id,
        min(order_date) as first_order,
        max(order_date) as most_recent_order,
        count(order_id) as number_of_orders
    from temp_orders
    group by customer_id

""")
ch2.createOrReplaceTempView("v_customer_orders")

pyspark

ch3 = spark.sql("""
    select
        orders.customer_id,
        sum(amount) as total_amount

    from raw_payments as payments 

    left join temp_orders as orders on
         payments.order_id = orders.order_id

    group by orders.customer_id
""")
ch3.createOrReplaceTempView("v_customer_payments")

sparksql


%%sql

select
    customers.id as customer_id,
    customers.first_name,
    customers.last_name,
    customer_orders.first_order,
    customer_orders.most_recent_order,
    customer_orders.number_of_orders,
    customer_payments.total_amount as customer_lifetime_value

from raw_customers as customers

left join v_customer_orders as customer_orders
    on customers.id = customer_orders.customer_id

left join v_customer_payments as customer_payments
    on  customers.id = customer_payments.customer_id

コードスニペット

pyspark の処理に困ったらコードスニペットの例も確認してみましょう。