1. ホーム
  2. python

[解決済み] sparkでデータフレームのカラムを更新する

2023-03-31 14:11:41

質問

新しいspark DataFrame APIを見ると、データフレームのカラムを変更できるかどうかが不明です。

行の値を変更するには、どのようにしたらよいでしょうか。 xy のようなデータフレームを作成できますか?

pandas であれば、これは

df.ix[x,y] = new_value

編集 : 下記で言われていることをまとめると、既存のデータフレームは不変なので修正できませんが、必要な修正を加えた新しいデータフレームを返すことはできます。

のように、ある条件に基づいてカラムの値を置き換えるだけであれば、そのカラムは np.where :

from pyspark.sql import functions as F

update_func = (F.when(F.col('update_col') == replace_val, new_value)
                .otherwise(F.col('update_col')))
df = df.withColumn('new_column_name', update_func)

カラムに対して何らかの操作を行い、データフレームに追加される新しいカラムを作成したい場合。

import pyspark.sql.functions as F
import pyspark.sql.types as T

def my_func(col):
    do stuff to column here
    return transformed_value

# if we assume that my_func returns a string
my_udf = F.UserDefinedFunction(my_func, T.StringType())

df = df.withColumn('new_column_name', my_udf('update_col'))

新しいカラムを古いカラムと同じ名前にしたい場合、追加のステップを追加することができます。

df = df.drop('update_col').withColumnRenamed('new_column_name', 'update_col')

どのように解決するのですか?

カラムをそのまま変更することはできませんが、カラムを操作して、その変更を反映した新しいDataFrameを返すことはできます。そのためには、まず UserDefinedFunction を作成し、その関数をターゲットとなるカラムにのみ選択的に適用します。Pythonの場合。

from pyspark.sql.functions import UserDefinedFunction
from pyspark.sql.types import StringType

name = 'target_column'
udf = UserDefinedFunction(lambda x: 'new_value', StringType())
new_df = old_df.select(*[udf(column).alias(name) if column == name else column for column in old_df.columns])

new_df と同じスキーマを持つようになりました。 old_df (と同じスキーマになります(仮に old_df.target_column は型が StringType のような型になります)しかし、カラム target_columnnew_value .