Ошибка несоответствия размеров в Spark ML

Question

Я довольно новичок как с ML, так и с Spark ML, и я пытаюсь создать модель прогнозирования с использованием нейронных сетей с Spark ML, но я получаю эту ошибку, когда вызываю метод .transform на моем научная модель. Проблема вызвана использованием OneHotEncoder, потому что без него все работает нормально. Я попытался вывести OneHotEncoder из трубопровода.

Мой вопрос: как я могу использовать OneHotEncoder и не получить эту ошибку?

java.lang.IllegalArgumentException: requirement failed: A & B Dimension mismatch! 
at scala.Predef$.require(Predef.scala:224)  at org.apache.spark.ml.ann.BreezeUtil$.dgemm(BreezeUtil.scala:41) at 
org.apache.spark.ml.ann.AffineLayerModel.eval(Layer.scala:163)  at 
org.apache.spark.ml.ann.FeedForwardModel.forward(Layer.scala:482) at 
org.apache.spark.ml.ann.FeedForwardModel.predict(Layer.scala:529) 

Мой код:

test_pandas_df = pd.read_csv(
    '/home/piotrek/ml/adults/adult.test', names=header, skipinitialspace=True) 
train_pandas_df = pd.read_csv(
    '/home/piotrek/ml/adults/adult.data', names=header, skipinitialspace=True) 
train_df = sqlContext.createDataFrame(train_pandas_df) 
test_df = sqlContext.createDataFrame(test_pandas_df) 

joined = train_df.union(test_df) 

assembler = VectorAssembler().setInputCols(features).setOutputCol("features") 

label_indexer = StringIndexer().setInputCol(
    "label").setOutputCol("label_index") 

label_indexer_fit = [label_indexer.fit(joined)] 

string_indexers = [StringIndexer().setInputCol(
    name).setOutputCol(name + "_index").fit(joined) for name in categorical_feats] 

one_hot_pipeline = Pipeline().setStages([OneHotEncoder().setInputCol(
    name + '_index').setOutputCol(name + '_one_hot') for name in categorical_feats]) 

mlp = MultilayerPerceptronClassifier().setLabelCol(label_indexer.getOutputCol()).setFeaturesCol(
    assembler.getOutputCol()).setLayers([len(features), 20, 10, 2]).setSeed(42L).setBlockSize(1000).setMaxIter(500) 
pipeline = Pipeline().setStages(label_indexer_fit 
           + string_indexers + [one_hot_pipeline] + [assembler, mlp]) 

model = pipeline.fit(train_df) 

# compute accuracy on the test set 
result = model.transform(test_df) 

## FAILS ON RESULT 

predictionAndLabels = result.select("prediction", "label_index") 

evaluator = MulticlassClassificationEvaluator(labelCol="label_index") 
print "-------------------------------" 
print("Test set accuracy = " + str(evaluator.evaluate(predictionAndLabels))) 
print "-------------------------------" 

Спасибо!

Answer 1

layersParam в модели не является правильным:

setLayers([len(features), 20, 10, 2]) 

Первый слой должен отражать количество функций ввода, которые в целом не будет таким же, как количество исходных столбцов перед кодированием.

Если вы не знаете общее количество функций спереди, вы можете, например, выделить отдельные функции и обучить модели. Псевдокод:

feature_pipeline_model = (Pipeline() 
    .setStages(...) # Only feature extraction 
    .fit(train_df)) 

train_df_features = feature_pipeline_model.transform(train_df) 
layers = [ 
    train_df_features.schema["features"].metadata["ml_attr"]["num_attrs"], 
    20, 10, 2 
]