Construct columns
A column is a logical construction that will be computed based on the data in a DataFrame using an expression
Construct a new column based on the input columns existing in a DataFrame
from pyspark.sql.functions import col
col("device")
df.device
df["device"] # 推荐，最通用好用
Use column objects to form complex expressions
col("ecommerce.purchase_revenue_in_usd") + col("ecommerce.total_item_quantity")
col("event_timestamp").desc()
(col("ecommerce.purchase_revenue_in_usd") * 100).cast("int")
Here’s an example of using these column expressions in the context of a DataFrame
recdf = (df.filter(col("ecommerce.purchase_revenue_in_usd").isNotNull())
.withColumn("purchase_revenue", (col("ecommerce.purchase_revenue_in_usd") * 100).cast("int"))
.withColumn("avg_purchase_revenue", col("ecommerce.purchase_revenue_in_usd") / col("ecommerce.total_item_quantity"))
.sort(col("avg_purchase_revenue").desc()))
display(revdf)
Subset columns
Use DataFrame transformations to subset columns
select
devicesDF = eventsDF.select("user_id", "device")
display(devicesDF)
from pyspark.sql.functions import col
locationsDF = eventsDF.select("user_id",
col("geo.city").alias("city"),
col("geo.state").alias("state"))
display(locationsDF)
selectExpr
appleDF = eventsDF.selectExpr("user_id", "device in ('macOS', 'iOS') as apple_user")
display(appleDF)
drop
Returns a new DataFrame after dropping the given column, specified as a string or column object
Use strings to specify multiple columns
anonymousDF = eventsDF.drop("user_id", "geo", "device")
noSalesDF = eventsDF.drop(col("ecommerce"))
Add or replace columns
Use DataFrame transformations to add or replace columns
withColumn最常用的方法
Returns a new DataFrame by adding a column or replacing the existing column that has the same name
mobileDF = df.withColumn("mobile", df["device"].isin("iOS", "Android"))
display(mobileDF)
purchaseQuantityDF = eventsDF.withColumn("purchase_quantity", col("ecommerce.total_item_quantity").cast("int"))
purchaseQuantityDF.printSchema()
withColumnRenamed 重命名
Returns a new DataFrame with a column renamed
locationDF = eventsDF.withColumnRenamed("geo", "location")
Subset Rows
Use DataFrame transformations to subset rows
purchasesDF = eventsDF.filter("ecommerce.total_item_quantity > 0")
revenueDF = eventsDF.filter(col("ecommerce.purchase_revenue_in_usd").isNotNull())
androidDF = eventsDF.filter((col("traffic_source") != "direct") & (col("device") == "Android"))
dropDuplicates
Returns a new DataFrame with duplicate rows removed, optionally considering only a subset of columns.
Alias: distinct
eventsDF.distinct() # 全字段去重
distinctUsersDF = eventsDF.dropDuplicates(["user_id"]) # 根据user_id去重
limit
limitDF = eventsDF.limit(100)
Sort Rows
sort()
Returns a new DataFrame sorted by the given columns or expressions.
Alias: orderBy
increaseTimestampsDF = eventsDF.sort("event_timestamp")
display(increaseTimestampsDF)
decreaseTimestampsDF = eventsDF.sort(col("event_timestamp").desc())
display(decreaseTimestampsDF)
increaseSessionsDF = eventsDF.orderBy(["user_first_touch_timestamp", "event_timestamp"])
display(increaseSessionsDF)
decreaseSessionsDF = eventsDF.sort(col("user_first_touch_timestamp").desc(), col("event_timestamp"))
display(decreaseSessionsDF)
