转换

Data Engineering 入门

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 列为两列
split_email = customer_df.email.str.split("@", expand=True)

# 此时，split_email 有两列：
# 第一列为 @ 前内容，第二列为 @ 后内容
# 用结果 DataFrame 创建两个新列。

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 !

Data Engineering 入门