DataFrameのColumnを列挙型で使用するにはどうすればいいですか

Question

DataFrameの特定の列にカテゴリ情報が含まれていると宣言するにはどうすればよいですか？

私はデータベースからロードしたSpark SQL DataFrameを持っています。このDataFrameの列の多くはカテゴリ情報を持っていますが、ロング（プライバシー用）としてエンコードされています。

私はspark-mlにこの列が数値であっても情報が実際にはCategoricalであることを伝えたいと思っています。カテゴリのインデックスにはいくつかの穴があり、それは受け入れられます。 （例：列の値が[1,0,0,4]の場合があります）

StringIndexerが存在することを知っていますが、特に多くの列があるため、エンコードとデコードの面倒を避けることをお勧めしますこの動作をしています。

私は、次の

train = load_from_database() 
categorical_cols = ["CategoricalColOfLongs1", 
        "CategoricalColOfLongs2"] 
numeric_cols = ["NumericColOfLongs1"] 

## This is what I am looking for 
## this step detects the min and max value of both columns 
## and adds metadata to indicate this as a categorical column 
## with (1 + max - min) categories 
categorizer = ColumnCategorizer(columns = categorical_cols, 
           autoDetectMinMax = True) 
## 

vectorizer = VectorAssembler(inputCols = categorical_cols + 
             numeric_cols, 
          outputCol = "features") 
classifier = DecisionTreeClassifier() 
pipeline = Pipeline(stages = [categorizer, vectorizer, classifier]) 
model = pipeline.fit(train) 

Answer 1

のように見える何かを探しされるだろう、私はあなたが本当に完全にこの問題を回避することはできません

、エンコードとデコードの手間を避けることを好むだろう。カテゴリ変数の必須メタデータは、実際には値とインデックスの間のマッピングです。それでも、手動で行うか、create a custom transformerにする必要はありません。

import numpy as np 
import pandas as pd 

df = sqlContext.createDataFrame(pd.DataFrame({ 
    "x1": np.random.random(1000), 
    "x2": np.random.choice(3, 1000), 
    "x4": np.random.choice(5, 1000) 
})) 

必要なのは、アセンブラとインデクサです：：

from pyspark.ml.feature import VectorAssembler, VectorIndexer 
from pyspark.ml import Pipeline 

pipeline = Pipeline(stages=[ 
    VectorAssembler(inputCols=df.columns, outputCol="features_raw"), 
    VectorIndexer(
     inputCol="features_raw", outputCol="features", maxCategories=10)]) 

transformed = pipeline.fit(df).transform(df) 
transformed.schema.fields[-1].metadata 

## {'ml_attr': {'attrs': {'nominal': [{'idx': 1, 
##  'name': 'x2', 
##  'ord': False, 
##  'vals': ['0.0', '1.0', '2.0']}, 
##  {'idx': 2, 
##  'name': 'x4', 
##  'ord': False, 
##  'vals': ['0.0', '1.0', '2.0', '3.0', '4.0']}], 
## 'numeric': [{'idx': 0, 'name': 'x1'}]}, 
## 'num_attrs': 3}} 

また、この例では、あなたがカテゴリとして、ベクトルの指定された要素をマークするために提供どのような種類の情報を示しますが、このようなデータフレームを持っていると仮定しましょうあなたは最初からこれを構築したい場合は、変数

{ 
    'idx': 2, # Index (position in vector) 
    'name': 'x4', # name 
    'ord': False, # is ordinal? 
    # Mapping between value and label 
    'vals': ['0.0', '1.0', '2.0', '3.0', '4.0'] 
} 

だから、あなたがしなければならないすべては、正しいスキーマです：

from pyspark.sql.types import * 
from pyspark.mllib.linalg import VectorUDT 

# Lets assume we have only a vector 
raw = transformed.select("features_raw") 

# Dictionary equivalent to transformed.schema.fields[-1].metadata shown abov 
meta = ... 
schema = StructType([StructField("features", VectorUDT(), metadata=meta)]) 

sqlContext.createDataFrame(raw.rdd, schema) 

しかし、必要なシリアライズ、デシリアライゼーションのために非常に非効率的です。あなたはまた、メタデータ引数を使用することができます

スパーク2.2ので：

df.withColumn("features", col("features").alias("features", metadata=meta)) 

は、私は、メタデータを見て、同じ技術を使用して、私はこのTransformerをコード化Attach metadata to vector column in Spark

Answer 2

ねえzero323も参照してください。

def _transform(self, data): 
    maxValues = self.getOrDefault(self.maxValues) 
    categoricalCols = self.getOrDefault(self.categoricalCols) 

    new_schema = types.StructType(data.schema.fields) 
    new_data = data 
    for (col, maxVal) in zip(categoricalCols, maxValues): 
     # I have not decided if I should make a new column or 
     # overwrite the original column 
     new_col_name = col + "_categorical" 

     new_data = new_data.withColumn(new_col_name, 
             data[col].astype(types.DoubleType())) 

     # metadata for a categorical column                                 
     meta = {u'ml_attr' : {u'vals' : [unicode(i) for i in range(maxVal + 1)], 
           u'type' : u'nominal', 
           u'name' : new_col_name}} 

     new_schema.add(new_col_name, types.DoubleType(), True, meta) 

    return data.sql_ctx.createDataFrame(new_data.rdd, new_schema)