高效的字符串后缀检测
Posted
技术标签:
【中文标题】高效的字符串后缀检测【英文标题】：Efficient string suffix detection 【发布时间】：2019-02-01 14:39:13 【问题描述】：我正在使用 PySpark 处理一个庞大的数据集，我想根据另一个数据帧中的字符串过滤数据帧。例如，
dd = spark.createDataFrame(["something.google.com","something.google.com.somethingelse.ac.uk","something.good.com.cy", "something.good.com.cy.mal.org"], StringType()).toDF('domains')
+----------------------------------------+
|domains |
+----------------------------------------+
|something.google.com |
|something.google.com.somethingelse.ac.uk|
|something.good.com.cy |
|something.good.com.cy.mal.org |
+----------------------------------------+
dd1 = spark.createDataFrame(["google.com", "good.com.cy"], StringType()).toDF('gooddomains')
+-----------+
|gooddomains|
+-----------+
|google.com |
|good.com.cy|
+-----------+
我假设domains 和gooddomains 是有效的域名。
我想要做的是过滤掉dd 中不以dd1 结尾的匹配字符串。所以在上面的例子中，我想过滤掉第 1 行和第 3 行，最终得到 p>
+----------------------------------------+
|domains |
+----------------------------------------+
|something.google.com.somethingelse.ac.uk|
|something.good.com.cy.mal.org |
+----------------------------------------+
我当前的解决方案（如下所示）最多只能考虑 3 个“单词”的域。如果我要在dd1（即白名单）中添加说verygood.co.ac.uk，那么它将失败。
def split_filter(x, whitelist):
splitted1 = x.select(F.split(x['domains'], '\.').alias('splitted_domains'))
last_two = splitted1.select(F.concat(splitted1.splitted_domains[F.size(splitted1.splitted_domains)-2], \
F.lit('.'), \
splitted1.splitted_domains[F.size(splitted1.splitted_domains)-1]).alias('last_two'))
last_three = splitted1.select(F.concat(splitted1.splitted_domains[F.size(splitted1.splitted_domains)-3], \
F.lit('.'), \
splitted1.splitted_domains[F.size(splitted1.splitted_domains)-2], \
F.lit('.'), \
splitted1.splitted_domains[F.size(splitted1.splitted_domains)-1]).alias('last_three'))
x = x.withColumn('id', F.monotonically_increasing_id())
last_two = last_two.withColumn('id', F.monotonically_increasing_id())
last_three = last_three.withColumn('id', F.monotonically_increasing_id())
final_d = x.join(last_two, ['id']).join(last_three, ['id'])
df1 = final_d.join(whitelist, final_d['last_two'] == whitelist['domains'], how = 'left_anti')
df2 = df1.join(whitelist, df1['last_three'] == whitelist['domains'], how = 'left_anti')
return df2.drop('id')
我正在使用 Spark 2.3.0 和 Python 2.7.5。
【问题讨论】：
你看过Efficient String Matching in Apache Spark吗？您也可以尝试使用正则表达式而不是udf。
是的，我看过那个链接。我不想要距离。我只需要过滤掉。
@Sotos 如果您有时间尝试编辑您的问题并在此处“我想要做的是过滤掉 dd 中不以 dd1 结尾的匹配字符串”部分更具体一点/跨度>
【参考方案1】：
让我们扩展 domains 以获得更好的覆盖范围：
domains = spark.createDataFrame([
"something.google.com", # OK
"something.google.com.somethingelse.ac.uk", # NOT OK
"something.good.com.cy", # OK
"something.good.com.cy.mal.org", # NOT OK
"something.bad.com.cy", # NOT OK
"omgalsogood.com.cy", # NOT OK
"good.com.cy", # OK
"sogood.example.com", # OK Match for shorter redundant, mismatch on longer
"notsoreal.googleecom" # NOT OK
], "string").toDF('domains')
good_domains = spark.createDataFrame([
"google.com", "good.com.cy", "alsogood.com.cy",
"good.example.com", "example.com" # Redundant case
], "string").toDF('gooddomains')
现在... 一个简单的解决方案，只使用 Spark SQL 原语，就是稍微简化您当前的方法。既然你说可以安全地假设这些是有效的公共域，我们可以定义一个这样的函数：
from pyspark.sql.functions import col, regexp_extract
def suffix(c):
return regexp_extract(c, "([^.]+\\.[^.]+$)", 1)
提取***域和一级子域：
domains_with_suffix = (domains
.withColumn("suffix", suffix("domains"))
.alias("domains"))
good_domains_with_suffix = (good_domains
.withColumn("suffix", suffix("gooddomains"))
.alias("good_domains"))
domains_with_suffix.show()
+--------------------+--------------------+
| domains| suffix|
+--------------------+--------------------+
|something.google.com| google.com|
|something.google....| ac.uk|
|something.good.co...| com.cy|
|something.good.co...| mal.org|
|something.bad.com.cy| com.cy|
| omgalsogood.com.cy| com.cy|
| good.com.cy| com.cy|
| sogood.example.com| example.com|
|notsoreal.googleecom|notsoreal.googleecom|
+--------------------+--------------------+
现在我们可以外连接了：
from pyspark.sql.functions import (
col, concat, lit, monotonically_increasing_id, sum as sum_
)
candidates = (domains_with_suffix
.join(
good_domains_with_suffix,
col("domains.suffix") == col("good_domains.suffix"),
"left"))
并过滤结果：
is_good_expr = (
col("good_domains.suffix").isNotNull() & # Match on suffix
(
# Exact match
(col("domains") == col("gooddomains")) |
# Subdomain match
col("domains").endswith(concat(lit("."), col("gooddomains")))
)
)
not_good_domains = (candidates
.groupBy("domains") # .groupBy("suffix", "domains") - see the discussion
.agg((sum_(is_good_expr.cast("integer")) > 0).alias("any_good"))
.filter(~col("any_good"))
.drop("any_good"))
not_good_domains.show(truncate=False)
+----------------------------------------+
|domains |
+----------------------------------------+
|omgalsogood.com.cy |
|notsoreal.googleecom |
|something.good.com.cy.mal.org |
|something.google.com.somethingelse.ac.uk|
|something.bad.com.cy |
+----------------------------------------+
这比Cartesian product required for direct join with LIKE 好，但无法满足蛮力，在最坏的情况下需要两次随机播放 - 一次用于join（如果good_domains 足够小到broadcasted，则可以跳过此操作)，另一个是group_by + agg。
不幸的是，Spark SQL 不允许自定义分区器对两者都只使用一个 shuffle（但是在 RDD API 中使用 composite key 是可能的）并且优化器还不够聪明，无法优化 join(_, "key1") 和 .groupBy("key1", _)。
如果您可以接受一些误报，则可以进行概率分析。首先让我们构建概率计数器（这里使用bounter 并得到toolz 的少量帮助）
from pyspark.sql.functions import concat_ws, reverse, split
from bounter import bounter
from toolz.curried import identity, partition_all
# This is only for testing on toy examples, in practice use more realistic value
size_mb = 20
chunk_size = 100
def reverse_domain(c):
return concat_ws(".", reverse(split(c, "\\.")))
def merge(acc, xs):
acc.update(xs)
return acc
counter = sc.broadcast((good_domains
.select(reverse_domain("gooddomains"))
.rdd.flatMap(identity)
# Chunk data into groups so we reduce the number of update calls
.mapPartitions(partition_all(chunk_size))
# Use tree aggregate to reduce pressure on the driver,
# when number of partitions is large*
# You can use depth parameter for further tuning
.treeAggregate(bounter(need_iteration=False, size_mb=size_mb), merge, merge)))
接下来定义一个像这样的用户定义函数函数
from pyspark.sql.functions import pandas_udf, PandasUDFType
from toolz import accumulate
def is_good_counter(counter):
def is_good_(x):
return any(
x in counter.value
for x in accumulate(lambda x, y: ".".format(x, y), x.split("."))
)
@pandas_udf("boolean", PandasUDFType.SCALAR)
def _(xs):
return xs.apply(is_good_)
return _
并过滤domains:
domains.filter(
~is_good_counter(counter)(reverse_domain("domains"))
).show(truncate=False)
+----------------------------------------+
|domains |
+----------------------------------------+
|something.google.com.somethingelse.ac.uk|
|something.good.com.cy.mal.org |
|something.bad.com.cy |
|omgalsogood.com.cy |
|notsoreal.googleecom |
+----------------------------------------+
在 Scala 中这可以通过 bloomFilter 完成
import org.apache.spark.sql.Column
import org.apache.spark.sql.functions._
import org.apache.spark.util.sketch.BloomFilter
def reverseDomain(c: Column) = concat_ws(".", reverse(split(c, "\\.")))
val checker = good_domains.stat.bloomFilter(
// Adjust values depending on the data
reverseDomain($"gooddomains"), 1000, 0.001
)
def isGood(checker: BloomFilter) = udf((s: String) =>
s.split('.').toStream.scanLeft("")
case ("", x) => x
case (acc, x) => s"$acc.$x"
.tail.exists(checker mightContain _))
domains.filter(!isGood(checker)(reverseDomain($"domains"))).show(false)
+----------------------------------------+
|domains |
+----------------------------------------+
|something.google.com.somethingelse.ac.uk|
|something.good.com.cy.mal.org |
|something.bad.com.cy |
|omgalsogood.com.cy |
|notsoreal.googleecom |
+----------------------------------------+
如果需要，shouldn't be hard to call such code from Python。
由于近似性质，这可能仍不能完全令人满意。如果您需要准确的结果，您可以尝试利用数据的冗余特性，例如使用trie（这里使用datrie 实现）。
如果good_domains 相对较小，您可以创建一个模型，方法与概率变体类似：
import string
import datrie
def seq_op(acc, x):
acc[x] = True
return acc
def comb_op(acc1, acc2):
acc1.update(acc2)
return acc1
trie = sc.broadcast((good_domains
.select(reverse_domain("gooddomains"))
.rdd.flatMap(identity)
# string.printable is a bit excessive if you need standard domain
# and not enough if you allow internationalized domain names.
# In the latter case you'll have to adjust the `alphabet`
# or use different implementation of trie.
.treeAggregate(datrie.Trie(string.printable), seq_op, comb_op)))
定义用户定义函数：
def is_good_trie(trie):
def is_good_(x):
if not x:
return False
else:
return any(
x == match or x[len(match)] == "."
for match in trie.value.iter_prefixes(x)
)
@pandas_udf("boolean", PandasUDFType.SCALAR)
def _(xs):
return xs.apply(is_good_)
return _
并将其应用于数据：
domains.filter(
~is_good_trie(trie)(reverse_domain("domains"))
).show(truncate=False)
+----------------------------------------+
|domains |
+----------------------------------------+
|something.google.com.somethingelse.ac.uk|
|something.good.com.cy.mal.org |
|something.bad.com.cy |
|omgalsogood.com.cy |
|notsoreal.googleecom |
+----------------------------------------+
这种特定方法的工作假设是所有good_domains 都可以压缩到一个单一的树中，但可以很容易地扩展以处理不满足此假设的情况。例如，您可以为每个***域或后缀（如在朴素解决方案中定义的那样）构建单个 trie
(good_domains
.select(suffix("gooddomains"), reverse_domain("gooddomains"))
.rdd
.aggregateByKey(datrie.Trie(string.printable), seq_op, comb_op))
然后，要么从序列化版本按需加载模型，要么使用RDD 操作。
这两种非原生方法可以根据数据、业务需求（如近似解的假阴性容忍度）和可用资源（驱动程序内存、执行程序内存、suffixes 的基数、访问分布式POSIX 兼容的分布式文件系统等）。在将这些应用于DataFrames 和RDDs（内存使用、通信和序列化开销）之间进行选择时，还需要考虑一些权衡。
* 见Understanding treeReduce() in Spark
【讨论】：
非常感谢。它有效并且提高了效率。也感谢您提供的所有信息。 这真的很棒。 +1【参考方案2】：如果我理解正确，您只需要使用简单的 SQL 字符串匹配模式的左反连接。
from pyspark.sql.functions import expr
dd.alias("l")\
.join(
dd1.alias("r"),
on=expr("l.domains LIKE concat('%', r.gooddomains)"),
how="leftanti"
)\
.select("l.*")\
.show(truncate=False)
#+----------------------------------------+
#|domains |
#+----------------------------------------+
#|something.google.com.somethingelse.ac.uk|
#|something.good.com.cy.mal.org |
#+----------------------------------------+
表达式concat('%', r.gooddomains) 在r.gooddomains 前面添加一个通配符。
接下来，我们使用l.domains LIKE concat('%', r.gooddomains) 来查找与此模式匹配的行。
最后，指定how="leftanti" 以便只保留不匹配的行。
更新：正如@user10938362 在the comments 中指出的那样，这种方法有两个缺陷：
1) 由于这只查看匹配的后缀，因此在某些极端情况下会产生错误的结果。例如：
example.com应该匹配example.com和subdomain.example.com，但不匹配fakeexample.com
有两种方法可以解决这个问题。首先是修改LIKE 表达式来处理这个问题。由于我们知道这些都是有效的域，我们可以检查是否完全匹配或域后跟一个点：
like_expr = " OR ".join(
[
"(l.domains = r.gooddomains)",
"(l.domains LIKE concat('%.', r.gooddomains))"
]
)
dd.alias("l")\
.join(
dd1.alias("r"),
on=expr(like_expr),
how="leftanti"
)\
.select("l.*")\
.show(truncate=False)
同样，可以将RLIKE 与带有后视功能的正则表达式模式一起使用。
2) 更大的问题是，正如here 所解释的，加入LIKE 表达式将导致笛卡尔积。如果dd1 小到可以广播，那么这不是问题。
否则，您可能会遇到性能问题，必须尝试不同的方法。
更多关于 Apache HIVE docs 中的 PySparkSQL LIKE 运算符：
A LIKE B:
如果字符串 A 匹配 SQL 简单正则表达式 B，则为 TRUE，否则为 FALSE。比较是逐个字符进行的。 B 中的
_字符匹配A 中的任意字符（类似于posix 正则表达式中的.），B 中的%字符匹配A 中任意数量的字符（类似于posix 正则表达式中的.*）。例如，'foobar' LIKE 'foo'评估为 FALSE，而'foobar' LIKE 'foo___'评估为 TRUE，'foobar' LIKE 'foo%'也是如此。要转义%，请使用\（%匹配一个%字符）。如果数据中包含分号，要搜索，需要转义，columnValue LIKE 'a\;b'
注意：这利用了使用pyspark.sql.functions.expr 到pass in a column value as a parameter to a function 的“技巧”。
【讨论】：
以上是关于高效的字符串后缀检测的主要内容，如果未能解决你的问题，请参考以下文章