Как проводить очистку данных с помощью Spark-Python на основе HDFS

Question

В настоящее время я концентрируюсь на предварительной обработке данных в проекте Data Mining. Если быть точным, я хочу выполнить очистку данных с помощью PySpark на основе HDFS. Я очень новичок в этих вещах, поэтому я хочу спросить, как это сделать?

Например, в HDFS есть таблица, содержащая следующие записи:

attrA   attrB   attrC      label
1       a       abc        0
2               abc        0
4       b       abc        1
4       b       abc        1
5       a       abc        0

После очистки всех записей, строка 2 <2, , abc, 0> должно иметь значение по умолчанию или вмененное значение для attrB, а строки 3 или 3 должны быть исключены. Так как я могу реализовать это с PySpark?

Answer 1

Исходя из того, что вы просили, есть две вещи, которые вы хотите достичь, сначала удалите дублирующиеся строки, которые могут быть достигнуты с помощью отдельной функции

df2 = df.distinct().show()

даст вам отдельные строки данных.

Второе - вменение пропущенных значений, что может быть достигнуто с помощью функции fillna.

df2 = df.na.fill({'attrB': 'm'}).show()

Answer 2

Это очень распространенная проблема в любом решении, управляемом данными. Лучший инструмент, который я могу порекомендовать для очистки данных с помощью Pyspark, - это Optimus.

Итак, давайте посмотрим. Сначала давайте предположим, что у вас уже есть этот DF в памяти:

df.show()

+-----+-----+-----+-----+
|attrA|attrB|attrC|label|
+-----+-----+-----+-----+
|    1|    a|  abc|    0|
|    2|     |  abc|    0|
|    4|    b|  abc|    1|
|    4|    b|  abc|    1|
|    5|    a|  abc|    0|
+-----+-----+-----+-----+

Для начала давайте создадим экземпляр DFT-трансфера:

transformer = op.DataFrameTransformer(df)

1. Установите значение по умолчанию для пустых ячеек:

df_default = transformer.replace_col(search='', change_to='new_value', columns='attrB').df

df_default.show()

+-----+---------+-----+-----+
|attrA|    attrB|attrC|label|
+-----+---------+-----+-----+
|    1|        a|  abc|    0|
|    2|new_value|  abc|    0|
|    4|        b|  abc|    1|
|    4|        b|  abc|    1|
|    5|        a|  abc|    0|
+-----+---------+-----+-----+

2. Исключить дубликаты записей:

df_clean = transformer.remove_duplicates(["attrA","attrB"]).dfdf_clean.show()

 +-----+---------+-----+-----+
 |attrA|    attrB|attrC|label|
 +-----+---------+-----+-----+
 |    4|        b|  abc|    1|
 |    5|        a|  abc|    0|
 |    1|        a|  abc|    0|
 |    2|new_value|  abc|    0|
 +-----+---------+-----+-----+