如何设置pyspark中不同表中的列值?
问题描述:
在Pyspark中 - 如何设置表A中column(listed_1)
的列值,其值为Table B (list_date)
,其值为where condition (B.list_expire_value) > 5 || (B.list_date) < 6
。 (B)表明它们是表B的列。如何设置pyspark中不同表中的列值?
目前我做的:
spark_df = table_1.join("table_2", on ="uuid").when((table_2['list_expire_value'] > 5) | (table_2['list_date'] < 6)).withColumn("listed_1", table_2['list_date'])
但我得到一个错误。这个怎么做?
Sample table : Table A uuid listed_1 001 abc 002 def 003 ghi Table B uuid list_date list_expire_value col4 001 12 7 dckvfd 002 14 3 dfdfgi 003 3 8 sdfgds Expected Output uuid listed1 list_expire_value col4 001 12 7 dckvfd 002 def 3 dfdfgi 003 3 8 sdfgds 002 of listed1 will not be replaced since they do not fufil the when conditions.
答
正确的形式是
from pyspark.sql import functions as F
spark_df = table_1.join(table_2, 'uuid', 'inner').withColumn('list_expire_value',F.when((table_2.list_expire_value > 5) | (table_2.list_date < 6), table_1.listed_1).otherwise(table_2.list_date)).drop(table_1.listed_1)
答
希望这有助于!
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
A = sc.parallelize([('001','abc'),('002','def'),('003','ghi')]).toDF(['uuid','listed_1'])
B = sc.parallelize([('001',12,7,'dckvfd'),('002',14,3,'dfdfgi'),('003',3,8,'sdfgds')]).\
toDF(['uuid','list_date','list_expire_value','col4'])
def cond_fn(x, y, z):
if (x > 5 or y < 6):
return y
else:
return z
final_df = A.join(B, on="uuid")
udf_val = udf(cond_fn, StringType())
final_df = final_df.withColumn("listed1",udf_val(final_df.list_expire_value,final_df.list_date, final_df.listed_1))
final_df.select(["uuid","listed1","list_expire_value","col4"]).show()
不要忘了让我们知道是否能解决你的问题:) pyspark SQL查询
@mtoto添加了产量预期。 – Viv
@tbone,用sqlContext,它会变成Update语句,设置col值= x。这是不允许的火花权利? – Viv
不,只是创建一个新的数据帧是SQL连接的结果 – tbone