ストリングフィールドを持つスパークのデシジョンツリー分類子にデータフレームを使用

Question

私はRDDベースのAPI用にデシジョンツリー分類子を使用することができましたが、今はSparkのDataframesベースのAPIに切り替えることを試みています。

私はこのようなデータセット（しかし、より多くのフィールドを持つ）を持つ：

国、目的地、期間、ラベル

Belgium, France, 10, 0 
Bosnia, USA, 120, 1 
Germany, Spain, 30, 0 

まず、私はデータフレームで私のcsvファイルを読み込む：

val data = session.read 
    .format("org.apache.spark.csv") 
    .option("header", "true") 
    .csv("/home/Datasets/data/dataset.csv") 

を

次に、文字列を数値列に変換します。

val stringColumns = Array("country", "destination") 

val index_transformers = stringColumns.map(
    cname => new StringIndexer() 
    .setInputCol(cname) 
    .setOutputCol(s"${cname}_index") 
) 

それから私はこのように、VectorAssemblerを使用して、1つのベクターに私のすべての機能を組み立てる：

val assembler = new VectorAssembler() 
    .setInputCols(Array("country_index", "destination_index", "duration_index")) 
    .setOutputCol("features") 

私はトレーニングやテストに私のデータを分割：

val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3)) 

その後、私は私のDecisionTree分類子を作成

val dt = new DecisionTreeClassifier() 
    .setLabelCol("label") 
    .setFeaturesCol("features") 

次に、パイプラインを使用してすべての変換を行います。

私は理解していない

val model = pipeline.fit(trainingData) 

val predictions = model.transform(testData) 

しかし、私はいくつかのミスを得る：

val pipeline = new Pipeline() 
    .setStages(Array(index_transformers, assembler, dt)) 

私は私のモデルを訓練し、予測のためにそれを使用

私はそのように私のコードを実行すると、私はこのエラーを持っています：

[error] found : Array[org.apache.spark.ml.feature.StringIndexer] 
[error] required: org.apache.spark.ml.PipelineStage 
[error]   .setStages(Array(index_transformers, assembler,dt)) 

だから何私がやったことは、私は右index_transformers valの後、右ヴァルアセンブラの前にパイプラインを追加したことである：

val index_pipeline = new Pipeline().setStages(index_transformers) 
val index_model = index_pipeline.fit(data) 
val df_indexed = index_model.transform(data) 

と私はトレーニングセットとテストセット、私の新しいdf_indexedデータフレームとして使用し、私はアセンブラと私のパイプラインからindex_transformersを削除し、DT

val Array(trainingData, testData) = df_indexed.randomSplit(Array(0.7, 0.3)) 

val pipeline = new Pipeline() 
    .setStages(Array(assembler,dt)) 

そして、私はこのエラーを取得する：

Exception in thread "main" java.lang.IllegalArgumentException: Data type StringType is not supported. 

これは基本的にStringのVectorAssemblerを使用していますが、df_indexedでは数値column_indexを使用するように言いましたが、vectorAssemblerでは使用していないようです。そして..

はあなたに

EDIT

ありがとう今、私はほとんどそれが仕事を得るために管理している：

value labels is not a member of org.apache.spark.ml.feature.StringIndexer 

：ことを除いて

val data = session.read 
    .format("org.apache.spark.csv") 
    .option("header", "true") 
    .csv("/home/hvfd8529/Datasets/dataOINIS/dataset.csv") 

val stringColumns = Array("country_index", "destination_index", "duration_index") 

val stringColumns_index = stringColumns.map(c => s"${c}_index") 

val index_transformers = stringColumns.map(
    cname => new StringIndexer() 
    .setInputCol(cname) 
    .setOutputCol(s"${cname}_index") 
) 

val assembler = new VectorAssembler() 
    .setInputCols(stringColumns_index) 
    .setOutputCol("features") 

val labelIndexer = new StringIndexer() 
    .setInputCol("label") 
    .setOutputCol("indexedLabel") 

val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3)) 

// Train a DecisionTree model. 
val dt = new DecisionTreeClassifier() 
    .setLabelCol("indexedLabel") 
    .setFeaturesCol("features") 
    .setImpurity("entropy") 
    .setMaxBins(1000) 
    .setMaxDepth(15) 

// Convert indexed labels back to original labels. 
val labelConverter = new IndexToString() 
    .setInputCol("prediction") 
    .setOutputCol("predictedLabel") 
    .setLabels(labelIndexer.labels()) 

val stages = index_transformers :+ assembler :+ labelIndexer :+ dt :+ labelConverter 

val pipeline = new Pipeline() 
    .setStages(stages) 


// Train model. This also runs the indexers. 
val model = pipeline.fit(trainingData) 

// Make predictions. 
val predictions = model.transform(testData) 

// Select example rows to display. 
predictions.select("predictedLabel", "label", "indexedFeatures").show(5) 

// Select (prediction, true label) and compute test error. 
val evaluator = new MulticlassClassificationEvaluator() 
    .setLabelCol("indexedLabel") 
    .setPredictionCol("prediction") 
    .setMetricName("accuracy") 
val accuracy = evaluator.evaluate(predictions) 
println("accuracy = " + accuracy) 

val treeModel = model.stages(2).asInstanceOf[DecisionTreeClassificationModel] 
println("Learned classification tree model:\n" + treeModel.toDebugString) 

今私はこれを言っエラーを持っています

と私は理解していない、私が従っているように（データラベルを持つ私の第二の問題について

val stages = index_transformers :+ assembler :+ labelIndexer :+ rf :+ labelConverter 

val pipeline = new Pipeline() 
    .setStages(stages) 

、私は.fitを使用するために必要な：スパークドキュメント上の例は：/

Answer 1

は、 ）この

val labelIndexer = new StringIndexer() 
    .setInputCol("label_fraude") 
    .setOutputCol("indexedLabel") 
    .fit(data) 

Answer 2

は次のようになります。私は私の最初の問題のためにした何

val pipeline = new Pipeline() 
    .setStages(index_transformers ++ Array(assembler, dt): Array[PipelineStage])