Spark数据帧更新列,其他列就像PySpark一样

Question

这将创建我的示例数据框:

df = sc.parallelize([('abc',),('def',)]).toDF() #(
df = df.selectExpr("_1 as one",)
df = df.withColumn("two", lit('z'))
df.show()

看起来像这样:

+---+---+
|one|two|
+---+---+
|abc|  z|
|def|  z|
+---+---+

现在我想要做的是一系列SQL where like语句,其中two无论匹配是否匹配列

在"伪代码"中它看起来像这样:

for letter in ['a','b','c','d']:
    df = df['two'].where(col('one').like("%{}%".format(letter))) += letter

最终导致df看起来像这样:

+---+----+
|one| two|
+---+----+
|abc|zabc|
|def|  zd|
+---+----+

Answer 1

如果您使用字符串列表来对字符串列进行子集化,则最好使用broadcast变量.让我们从一个更现实的例子开始,你的字符串仍然包含空格:

df = sc.parallelize([('a b c',),('d e f',)]).toDF()
df = df.selectExpr("_1 as one",)
df = df.withColumn("two", lit('z'))

然后我们broadcast从一个字母列表中创建一个变量,然后定义一个udf使用它们来对字符串列表进行子集化的变量.最后用另一列中的值连接它们,返回一个字符串:

letters = ['a','b','c','d']
letters_bd = sc.broadcast(letters) 

def subs(col1, col2):

    l_subset =  [x for x in col1 if x in letters_bd.value]
    return col2 + ' ' + ' '.join(l_subset)

subs_udf = udf(subs)

要应用上述内容,我们要进行子集化的字符串需要转换为列表,因此我们split()首先使用该函数,然后应用我们的udf:

from pyspark.sql.functions import col, split

df.withColumn("three", split(col('one'), r'\W+')) \
  .withColumn("three", subs_udf("three", "two")) \
  .show()
+-----+---+-------+
|  one|two|  three|
+-----+---+-------+
|a b c|  z|z a b c|
|d e f|  z|    z d|
+-----+---+-------+

或者没有udf,使用regexp_replace和concat如果您的字母可以舒适地融入regex表达式.

from pyspark.sql.functions import regexp_replace, col, concat, lit

df.withColumn("three", concat(col('two'), lit(' '), 
              regexp_replace(col('one'), '[^abcd]', ' ')))