Функция уменьшения Pyspark вызывает StackOverflowError
Я работаю с довольно большим кадром данных (около 100 тысяч строк с намерением достичь 10 миллионов), и он имеет следующую структуру:
+------+--------------------+--------+--------------------+-------------------+
|LineId| Content| EventId| EventTemplate| Timestamp|
+------+--------------------+--------+--------------------+-------------------+
| 1|Receiving block b...|ef6f4915|Receiving block <...|2009-11-08 20:35:18|
| 2|BLOCK* NameSystem...|9bc09482|BLOCK* NameSystem...|2009-11-08 20:35:18|
| 3|Receiving block b...|9ca53bce|Receiving block <...|2009-11-08 20:35:19|
+------+--------------------+--------+--------------------+-------------------+
Я хотел бы добавить метку, и я использую для этого следующую функцию:
from functools import reduce
label_condition = reduce(lambda a, b: a|b, (df['Content'].like('%'+pat+"%") for pat in blocks))
где blocks — список, содержащий блок (назовем его токеном), определяющий, является ли строка аномальной. Эта функция проверяет, содержит ли поле Content какое-либо значение из списка blocks.
Размер этого списка составляет около 17 КБ, что, я думаю, и является причиной проблемы.
Когда я пытаюсь добавить это в фрейм данных или просто оценить эту операцию, я получаю следующую ошибку:
Py4JJavaError: An error occurred while calling o50581.toString.
: java.lang.StackOverflowError
at org.apache.spark.sql.catalyst.util.package$$anonfun$usePrettyExpression$1.applyOrElse(package.scala:128)
at org.apache.spark.sql.catalyst.util.package$$anonfun$usePrettyExpression$1.applyOrElse(package.scala:128)
at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDown$1(TreeNode.scala:318)
...
Посмотрев в Интернете, я увидел, что это может быть проблема с выполнением слишком сложного плана из Spark и / или с использованием контрольной точки, чтобы избежать подобных вещей, но я не уверен, как это сделать. Я попытался добавить контрольную точку перед оценкой этого, а также попытался использовать выбор, чтобы уменьшить df только до столбца «Содержание», но безрезультатно.
Я нашел это решение в Scala для оптимизации функции редукции, но я не знаю, как его перевести для python.
Есть ли способ оптимизировать это или сделать, по крайней мере, шаг за шагом или итеративно, чтобы избежать переполнения стека?
Заранее спасибо.
В мелком масштабе работает отлично. Если вам нужен пример, вы можете проверить ответ на мой вопрос пару часов назад, в котором объясняется функция.
сколько элементов в blocks? ошибка возникает при большом количестве рекурсий, которые в вашем случае являются условиями ИЛИ.
В списке blocks ровно 16838 элементов.
попробуйте использовать метод rlike
Только что попробовал. Та же ошибка, к сожалению. @самкарт
withColumn('label', func.col('log').rlike('|'.join(anomalous_blocks))) -- пробовали? rlike принимает регулярное выражение.
это withColumn('label', func.col('log').rlike('|'.join(anomalous_blocks))) сработало. Спасибо!!!! Если вы напишете это как ответ, я буду считать, что это решено @samkart
Ответы 1
вы можете попробовать использовать метод rlike, который принимает регулярное выражение - передайте шаблон регулярного выражения как 'element1|element2|...'.
data_sdf. \
withColumn('label', func.col('log').rlike('|'.join(anomalous_blocks))). \
show()
# +---+---------------+-----+
# | id| log|label|
# +---+---------------+-----+
# | 1|Test logX blk_A| true|
# | 2|Test logV blk_B|false|
# | 3|Test logF blk_D| true|
# | 4|Test logD blk_F|false|
# | 5|Test logB blk_K|false|
# | 6|Test logY blk_A| true|
# | 7|Test logE blk_C| true|
# +---+---------------+-----+
Можете ли вы проверить, что происходит, если вы просто протестируете в меньшем масштабе, поэтому давайте попробуем, например, с блоками, состоящими только из двух элементов?