PySpark: выполнение одной и той же операции, несколько столбцов

Question

Я довольно новичок в Spark / PySpark и пытаюсь быть последовательным в следовании «лучшим практикам», но, к сожалению, иногда я возвращаюсь к хакерским решениям, как показано ниже.

Я работаю над набором данных твита который содержит базовые твиты, ретвитированный контент и цитируемый контент. Если твит не является retweeted_status или quoted_status, каждая подфункция для этого имеет значение Нет или пустой список или какой-либо другой вариант в зависимости от типа данных.

Я пытаюсь создать новые столбцы для каждого из эти функции, используя контент из базовой функции, ЕСЛИ это не ретвит или цитируемый статус, ИЛИ не использовать ретвитированный контент, ИЛИ не использовать базовую функцию + цитируемый контент.

def _shared_content(feature, df):
    return df.withColumn(f'tweet_{feature.replace("entities.", "")}',
        when((col(f'`retweeted_status.{feature}`').isNotNull()),
              df[f'`retweeted_status.{feature}`'])
        .when((col(f'`quoted_status.{feature}`').isNotNull()),
                concat(
                       df[f'`{feature}`'],
                       lit(" "),
                       df[f'`quoted_status.{feature}`']))
        .otherwise(df[f'`{feature}`']))

common_features = [
    'text',
    'entities.hashtags',
    'entities.media',
    'entities.urls',
]

for f in common_features:
    df = df.transform(lambda df, f=f: _shared_content(f, df))

Как вы видно, что это немного беспорядок, поэтому я написал немного псевдокода для разборчивости. Здесь я выполняю следующие функции:

• Для каждой общей функции:
  • Если retweet_status. [FEATURE] не равно None, установите для нового col tweet_ [FEATURE] значение retweet_status . [FEATURE]
  • Если в качестве кавычки указано [FEATURE] значение None, установите для нового col tweet_ [FEATURE] значение [FEATURE] + "" + quoted_status. [FEATURE]
  • В противном случае установите tweet_ [ОСОБЕННОСТИ] для базы [ОСОБЕННОСТИ].

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

В качестве последнего пояснения я выполняю то же преобразование, но единственное, что изменяет, - это функция, над которой я работаю

РЕДАКТИРОВАТЬ: Я сделал это немного более разборчиво с:

def _shared_content(f, df):
    new_col = f'tweet_{f.replace("entities.", "")}'

    retweet_cond = ((
        col(f'`retweeted_status.{f}`').isNotNull()),
                    df[f'`retweeted_status.{f}`'])
    quoted_cond = ((
        col(f'`quoted_status.{f}`').isNotNull()),
                   concat(df[f'`{f}`'], lit(" "), df[f'`quoted_status.{f}`']))

    return df.withColumn(
        new_col,
        when(*retweet_cond)
        .when(*quoted_cond)
        .otherwise(df[f'`{f}`'])
    )

Answer 1

Я бы написал что-то вроде этого:

def _shared_content(feature, df):
    feat_col = col(feature)
    retweet = col(f"retweeted_status.{feature}")
    quoted = col(f"quoted_status.{feature}")
    new_feat_name = f'tweet_{feature.replace("entities.", "")}'

    return df.withColumn(
        new_feat_name,
        (
            when(retweet.isNotNull(), retweet)
            .when(quoted.isNotNull(), concat(feat_col, lit(" "), quoted))
            .otherwise(feat_col)
        ),
    )

Некоторые принципы, которым я обычно следую, когда пишу код с помощью Pyspark (или чего-либо еще):

• Избегайте дублирования кода (вы повторяли имена столбцов в нескольких местах)
• Замена необработанных значений именами переменных, когда это помогает удобочитаемости (как в случае с новым именем объекта)
• Использование объектов столбца Spark вместо ненужной ссылки тот же самый DataFrame с df["<column name>"].

PS. Я не уверен, почему вы используете обратные кавычки.