26

Spark でデータを読み込んで処理するために使用している Python クラスがあります。やらなければならないさまざまなことの中で、Spark データフレームのさまざまな列から派生したダミー変数のリストを生成しています。私の問題は、必要なことを達成するためにユーザー定義関数を適切に定義する方法がわからないことです。

私は現在、基礎となるデータフレームRDDにマップされたときに問題の半分を解決するメソッドを持っています(これはより大きなdata_processorクラスのメソッドであることを思い出してください):

def build_feature_arr(self,table):
    # this dict has keys for all the columns for which I need dummy coding
    categories = {'gender':['1','2'], ..}

    # there are actually two differnt dataframes that I need to do this for, this just specifies which I'm looking at, and grabs the relevant features from a config file
    if table == 'users':
        iter_over = self.config.dyadic_features_to_include
    elif table == 'activty':
        iter_over = self.config.user_features_to_include

    def _build_feature_arr(row):
        result = []
        row = row.asDict()
        for col in iter_over:
            column_value = str(row[col]).lower()
            cats = categories[col]
            result += [1 if column_value and cat==column_value else 0 for cat in cats]
        return result
    return _build_feature_arr

基本的に、これが行うことは、指定されたデータフレームに対して、指定された列のカテゴリ変数値を取得し、これらの新しいダミー変数の値のリストを返すことです。つまり、次のコードです。

data = data_processor(init_args)
result = data.user_data.rdd.map(self.build_feature_arr('users'))

次のようなものを返します。

In [39]: result.take(10)
Out[39]:
[[1, 0, 0, 0, 1, 0],
 [1, 0, 0, 1, 0, 0],
 [1, 0, 0, 0, 0, 0],
 [1, 0, 1, 0, 0, 0],
 [1, 0, 0, 1, 0, 0],
 [1, 0, 0, 1, 0, 0],
 [0, 1, 1, 0, 0, 0],
 [1, 0, 1, 1, 0, 0],
 [1, 0, 0, 1, 0, 0],
 [1, 0, 0, 0, 0, 1]]

これはまさに、必要なダミー変数のリストを生成するという点で私が望んでいることですが、ここに私の質問があります: (a) Spark SQL クエリで使用できる同様の機能を持つ UDF を作成するにはどうすればよいですか (または他の方法で) 、私は推測します)、または(b)上記のマップから得られたRDDを取得し、それをuser_dataデータフレームに新しい列として追加しますか?

feature_arrayいずれにせよ、私がする必要があるのは、上記の関数の出力 (または機能的に同等のもの) を含む新しい列 (と呼びましょう) とともに、user_data の列を含む新しいデータフレームを生成することです。

4

1 に答える 1

43

スパーク >= 2.3、>= 3.0

Spark 2.3OneHotEncoderは推奨されていないため、OneHotEncoderEstimator. 最近のリリースを使用する場合は、encoderコードを変更してください

from pyspark.ml.feature import OneHotEncoderEstimator

encoder = OneHotEncoderEstimator(
    inputCols=["gender_numeric"],  
    outputCols=["gender_vector"]
)

Spark 3.0 では、このバリアントは次のように名前が変更されましたOneHotEncoder。

from pyspark.ml.feature import OneHotEncoder

encoder = OneHotEncoder(
    inputCols=["gender_numeric"],  
    outputCols=["gender_vector"]
)

さらにStringIndexer、複数の入力列をサポートするように拡張されました。

StringIndexer(inputCols=["gender"], outputCols=["gender_numeric"])

スパーク < 2.3

UDF を作成することはできますが、なぜ作成するのでしょうか? このカテゴリのタスクを処理するために設計されたかなりの数のツールが既にあります。

from pyspark.sql import Row
from pyspark.ml.linalg import DenseVector

row = Row("gender", "foo", "bar")

df = sc.parallelize([
  row("0", 3.0, DenseVector([0, 2.1, 1.0])),
  row("1", 1.0, DenseVector([0, 1.1, 1.0])),
  row("1", -1.0, DenseVector([0, 3.4, 0.0])),
  row("0", -3.0, DenseVector([0, 4.1, 0.0]))
]).toDF()

まず第一にStringIndexer。

from pyspark.ml.feature import StringIndexer

indexer = StringIndexer(inputCol="gender", outputCol="gender_numeric").fit(df)
indexed_df = indexer.transform(df)
indexed_df.drop("bar").show()

## +------+----+--------------+
## |gender| foo|gender_numeric|
## +------+----+--------------+
## |     0| 3.0|           0.0|
## |     1| 1.0|           1.0|
## |     1|-1.0|           1.0|
## |     0|-3.0|           0.0|
## +------+----+--------------+

次へOneHotEncoder：

from pyspark.ml.feature import OneHotEncoder

encoder = OneHotEncoder(inputCol="gender_numeric", outputCol="gender_vector")
encoded_df = encoder.transform(indexed_df)
encoded_df.drop("bar").show()

## +------+----+--------------+-------------+
## |gender| foo|gender_numeric|gender_vector|
## +------+----+--------------+-------------+
## |     0| 3.0|           0.0|(1,[0],[1.0])|
## |     1| 1.0|           1.0|    (1,[],[])|
## |     1|-1.0|           1.0|    (1,[],[])|
## |     0|-3.0|           0.0|(1,[0],[1.0])|
## +------+----+--------------+-------------+

VectorAssembler:

from pyspark.ml.feature import VectorAssembler

assembler = VectorAssembler(
    inputCols=["gender_vector", "bar", "foo"], outputCol="features")

encoded_df_with_indexed_bar = (vector_indexer
    .fit(encoded_df)
    .transform(encoded_df))

final_df = assembler.transform(encoded_df)

カテゴリ変数が含まれている場合は、必要なメタデータを設定するためにbar使用できます。VectorIndexer

from pyspark.ml.feature import VectorIndexer

vector_indexer = VectorIndexer(inputCol="bar", outputCol="bar_indexed")

しかし、ここではそうではありません。

最後に、パイプラインを使用してすべてをラップできます。

from pyspark.ml import Pipeline
pipeline = Pipeline(stages=[indexer, encoder, vector_indexer, assembler])
model = pipeline.fit(df)
transformed = model.transform(df)

おそらく、すべてをゼロから作成するよりも、はるかに堅牢でクリーンなアプローチです。特に、異なるデータセット間で一貫したエンコードが必要な場合は、いくつかの注意事項があります。StringIndexer詳細については、およびの公式ドキュメントを参照してVectorIndexerください。

同等の出力を取得する別の方法はRFormula whichです。

RFormula特徴のベクトル列と、ラベルの double または string 列を生成します。線形回帰のために R で数式が使用される場合と同様に、文字列入力列はワンホット エンコードされ、数値列は double にキャストされます。ラベル列が文字列型の場合、最初に で double に変換されStringIndexerます。ラベル列が DataFrame に存在しない場合、出力ラベル列は式で指定された応答変数から作成されます。

from pyspark.ml.feature import RFormula

rf = RFormula(formula="~ gender +  bar + foo - 1")
final_df_rf = rf.fit(df).transform(df)

ご覧のとおり、はるかに簡潔ですが、構成が難しいため、あまりカスタマイズできません。それにもかかわらず、このような単純なパイプラインの結果は同じになります:

final_df_rf.select("features").show(4, False)

## +----------------------+
## |features              |
## +----------------------+
## |[1.0,0.0,2.1,1.0,3.0] |
## |[0.0,0.0,1.1,1.0,1.0] |
## |(5,[2,4],[3.4,-1.0])  |
## |[1.0,0.0,4.1,0.0,-3.0]|
## +----------------------+


final_df.select("features").show(4, False)

## +----------------------+
## |features              |
## +----------------------+
## |[1.0,0.0,2.1,1.0,3.0] |
## |[0.0,0.0,1.1,1.0,1.0] |
## |(5,[2,4],[3.4,-1.0])  |
## |[1.0,0.0,4.1,0.0,-3.0]|
## +----------------------+

ご質問について：

Spark SQLクエリで使用できる同様の機能を持つUDFを作成します（または他の方法だと思います）

これは、他のものと同じように単なる UDF です。すべてが正常に機能するように、サポートされているタイプを使用していることを確認してください。

上記のマップから得られた RDD を取得し、それを新しい列として user_data データフレームに追加しますか?

from pyspark.ml.linalg import VectorUDT
from pyspark.sql.types import StructType, StructField

schema = StructType([StructField("features", VectorUDT(), True)])
row = Row("features")
result.map(lambda x: row(DenseVector(x))).toDF(schema)

注:

Spark 1.x の場合は に置き換えpyspark.ml.linalgますpyspark.mllib.linalg。

于 2015-10-07T05:59:07.113 に答える