変換

データエンジニアリング入門

Vincent Vankrunkelsven

Data Engineer @ DataCamp

変換の種類

customer_id	email	state	created_at
1	[email protected]	New York	2019-01-01 07:00:00

属性の選択（例: 'email'）
コード値の変換（例: 'New York' -> 'NY'）
データ検証（例: 'created_at' の日付入力）
列の分割（複数列に分ける）
複数ソースの結合

例: 分割（Pandas）

customer_id	email	username	domain
1	[email protected]	jane.doe	theweb.com

customer_df # 顧客データの Pandas DataFrame

# email 列を '@' で 2 列に分割
split_email = customer_df.email.str.split("@", expand=True)

# ここで split_email は 2 列になり、
# 1 列目に @ より前、2 列目に @ より後が入る
# 得られた DataFrame から 2 つの新列を作成

customer_df = customer_df.assign(
  username=split_email[0],
  domain=split_email[1],
)

PySpark での変換

PySpark へのデータ取り込み

import pyspark.sql

spark = pyspark.sql.SparkSession.builder.getOrCreate()


spark.read.jdbc("jdbc:postgresql://localhost:5432/pagila",

                "customer",

                properties={"user":"repl","password":"password"})

例: 結合

新しい ratings テーブル

customer_id	film_id	rating
1	2	1
2	1	5
2	2	3
...	...	...

customer テーブル

customer_id	first_name	last_name	...
1	Jane	Doe	...
2	Joe	Doe	...
...	...	...	...

customer_id は ratings テーブルと重複

例: 結合（PySpark）

customer_df # 顧客データの PySpark DataFrame
ratings_df # 評価データの PySpark DataFrame


# 評価を顧客単位で集計
ratings_per_customer = ratings_df.groupBy("customer_id").mean("rating")


# customer_id で結合
customer_df.join(
  ratings_per_customer,
  customer_df.customer_id==ratings_per_customer.customer_id
)

Passons à la pratique !

データエンジニアリング入門