将正则表达式应用于 spark 数据帧的每一行并将其另存为同一数据帧中的新列答案

【问题标题】：Apply regex to every row of a spark dataframe and save it as a new column in the same dataframe将正则表达式应用于 spark 数据帧的每一行并将其另存为同一数据帧中的新列
【发布时间】：2016-08-22 19:44:08
【问题描述】：

假设我有一个 spark 数据框，

data.show()

ID                URL
1             https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed
2             https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed
3             https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed
4             https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed
5                 None

我想为它编写一个正则表达式操作，我想在其中解析特定场景的 URL。场景将是在 &q 之后和下一个 & 之前解析。我可以在 python 中为 python 数据框编写这个，如下所示，

re.sub(r"\s+", " ", re.search(r'/?q=([^&]*)', data['url'][i]).group(1).replace('+', ' ')

我想在 pyspark 中写同样的内容。

如果写类似的东西，

 re.sub(r"\s+", " ", re.search(r'/?q=([^&]*)', data.select(data.url.alias("url")).collect()).group(1).replace('+', ' '))

或

re.sub(r"\s+", " ", re.search(r'/?q=([^&]*)', data.select(data['url']).collect()).group(1).replace('+', ' '))

我收到以下错误，

TypeError: expected string or buffer

一种选择是使用将数据转换为熊猫，

data.toPandas() 然后进行操作。但是我的数据很大，将其转换为 pandas 会使它变慢。有没有办法可以直接将其写入 spark 数据框中的新列，我可以喜欢，

ID                URL                                                                           word
1             https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed       To Be Parsed out   
2             https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed       To Be Parsed out 
3             https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed       To Be Parsed out 
4             https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed       To Be Parsed out 
5             None                                                                             None

我们如何才能将其添加为 pyspark 数据框中的新表？哪个适用于数据框的每一行？

【问题讨论】：

考虑使用 udf 和 withColumn 或将您的数据框转换为 rdd 并使用 map
@David 我是 pyspark 的新手。刚认识这些。如果您不介意，可以请您帮忙使用地图功能吗？

标签： python python-2.7 apache-spark pyspark pyspark-sql

【解决方案1】：

正如@David 在评论中提到的，您可以使用udf 和withColumn：

Scala 代码：

import org.apache.spark.sql.functions._

val getWord: (String => String) = (url: String) => {
  if (url != null) {
    """/?q=([^&]*)""".r
      .findFirstIn(url)
      .get
      .replaceAll("q=", "")
      .replaceAll("\\+", " ")
  }
  else
    null
}
val udfGetWord = udf(getWord)
df.withColumn("word", udfGetWord($"url")).show()

【讨论】：

【解决方案2】：

Pyspark 代码：

#Create dataframe with sample data

df = spark.createDataFrame([(1,'https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed'),(2,'https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed'),(3,'https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed'),(4,'https://www.sitename.com/&q=To+Be+Parsed+out&oq=Dont+Need+to+be+parsed'),(5,'None')],['id','url'])

使用substr 使用位置索引剪切所需的字符串，使用instr 识别搜索模式的位置。

regexp_replace 用于将“+”号替换为空格。

df.selectExpr("id",
              "url",
              "regexp_replace(substr(url,instr(url,'&q')+3, instr(url,'&oq') - instr(url,'&q') - 3 ),'\\\+',' ') AS word")\
.show()

#+---+--------------------+----------------+
#| id|                 url|            word|
#+---+--------------------+----------------+
#|  1|https://www.siten...|To Be Parsed out|
#|  2|https://www.siten...|To Be Parsed out|
#|  3|https://www.siten...|To Be Parsed out|
#|  4|https://www.siten...|To Be Parsed out|
#|  5|                None|                |
#+---+--------------------+----------------+

如果搜索字符串中不存在搜索模式，则返回空白。这可以使用case 语句处理。

【讨论】：