def column_concat(a,b):
return concat(a,b)
searches_df = searches_df.withColumn('unique_id',reduce(column_concat,(searches_df[col] for col in search_parameters)))
此方法有效,除非列包含空值,否则整个连接的字符串为空。我想要一个占位符或一些字符,而不是在连接字符串。
给定以下数据帧:
df = sqlContext.createDataFrame([("foo", "bar"), ("baz", None)],
('a', 'b'))
df.show()
+---+----+
| a| b|
+---+----+
|foo| bar|
|baz|null|
+---+----+
一个解决方案是使用udf来过滤/替换空,例如:
import pyspark.sql.functions as F
from pyspark.sql.types import StringType
concat_udf = F.udf(lambda cols: "".join([x if x is not None else "*" for x in cols]), StringType())
df.withColumn("unique_id", concat_udf(F.array("a", "b"))).show()
收益率:
+---+----+---------+
| a| b|unique_id|
+---+----+---------+
|foo| bar| foobar|
|baz|null| baz*|
+---+----+---------+
另外:
import pyspark.sql.functions as F
def myConcat(*cols):
return F.concat(*[F.coalesce(c, F.lit("*")) for c in cols])
df.withColumn("unique_id", myConcat("a", "b")).show()
也产生:
+---+----+---------+
| a| b|unique_id|
+---+----+---------+
|foo| bar| foobar|
|baz|null| baz*|
+---+----+---------+
df = df.withColumn('joined_column', sf.concat(sf.col('colname1'),sf.lit('_'), sf.col('colname2')))
首先需要像下面这样导入pyspark sql函数
from pyspark.sql import functions as sf
df = df.withColumn('joined_column', sf.concat(sf.col('colname1'),sf.lit('_'), sf.col('colname2')))
在Spark Scala中,您可以首先用NA
填充这些列的空值(可以取任何值),然后执行连接操作。它工作得很好。
df = df.na.fill("NA").concat(col A, col B, col C, col D, col E)