Проверьте, пуст ли фрейм данных pyspark, что вызывает проблемы с памятью

Question

У меня есть таблица, в которой около 1 миллиарда записей. Я запускаю запрос, чтобы найти дубликаты. Если результатом запроса является 0 строк, дубликатов нет, иначе есть. Если есть дубликаты, я хочу записать это имя таблицы в текстовый файл. Так что я делаю

df = spark.sql("SELECT count(*) FROM table GROUP BY key1,key2,key3 HAVING count(*) > 1 LIMIT 1)
if df.count() > 0:
    with open('duplicate_tables.txt','a') as file:
        file.write('\n' + table)

На df.count() строка, я получаю сообщение об ошибке java.io.IOException: No space left on device. Это потому чтоcount()неэффективно. Я также получаю ту же ошибку, когда пытаюсь использовать

if len(df.head(1)) != 0:

В своем запросе я подумал (надеялся), что добавление LIMIT 1 поможет, чтобы ему не пришлось проходить сотни и сотни строк, просто проверьте, пуст он или нет. Если я вынимаю счетную часть, все работает нормально.

Я видел несколько способов переписать оператор count (я прошел через Как проверить, пуст ли фрейм данных искры?), Но пока мне не повезло.

Answer 1

Спарк ленивый. Это означает, что когда вы бежитеspark.sql()на самом деле ничего не происходит. Вы можете убедиться в этом, заметив, чтоspark.sql()"выполняется" немедленно, независимо от сложности SQL. Фактическая обработка выполняется после того, как требуется действие; в вашем случае, когда.count()вступает в игру. Последнее, вероятно, вызывает проблему с памятью из-за сложности SQL и размера таблицы.

Возможно, вы могли бы попробовать еще одну вещь - прочитать всю таблицу и запустить Spark для проверки наличия дубликатов. Однако, учитывая исходный размер вашей таблицы, это также может привести к проблемам с памятью.

df = spark.sql("SELECT * FROM table") # or select particular column(s)
if df.count() != df.dropDuplicates().count():
    with open('duplicate_tables.txt','a') as file:
        file.write('\n' + table)

Answer 2

Я обнаружил, что это более эффективный способ определить, действительно ли фрейм данных искры пуст:

df.first() == Нет

после тестирования с помощью .limit(1), .head(), .count () и т. д.