スパークデータフレームの手段でヌル値を置換しようとしました。

Question

更新：間違っていました。エラーはランダムなフォレストではなくベクトルアセンブラに由来します。しかし、エラー/問題は同じです。 vectorAssemblerでdf_noNullsデータフレームを使用すると、ヌル値があるため列をベクトル化できないという。

私はこの質問の他の回答を見て、これを動作させるために回答コードを解放/借り/盗んだ。私の最終目標は、ヌル値をとらないRF/GB /その他のMLモデリングです。すべての数値列を取得し、各列の意味を取得し、2つの要素を結合し、すべてのNULLを平均で置き換える新しいデータフレームを作成するために、次のコードをまとめました。ランダムフォレストの「機能」部分として数値列のベクトルを作成しようとすると、「集計する値をnullにできません」というエラーが返されます。

val numCols = DF.schema.fields filter { 
x => x.dataType match { 
    case x: org.apache.spark.sql.types.DoubleType => true 
    case x: org.apache.spark.sql.types.IntegerType => true 
    case x: org.apache.spark.sql.types.LongType => true 
    case _ => false 
    } 
    } map {x => x.name} 
//NUMCOLS NOW IS AN ARRAY OF ALL NUMERIC COLUMN NAMES 
val numDf = DF.select(numCols.map(col): _*) 
//NUMDF IS A DATAFRAME OF ALL NUMERIC COLUMNS 

val means = numDf.agg(numDf.columns.map(c => (c -> "avg")).toMap) 
//CREATES A DATAFRAME OF MEANS OF ALL NUMERIC VARIABLES 
means.persist() 
//PERSIST TABLE 'MEANS' FOR JOINING --BROADCAST ALSO WORKS BUT I WAS GETTING MEMORY ISSUES WITH IT SO I SWITCHED IT 


val exprs = numDf.columns.map(c => coalesce(col(c), col(s"avg($c)")).alias(c)) 
//EXPRS CREATES FUNCTION TO REPLACE NULLS WITH MEANS 

val df_noNulls = DF.crossJoin(means).select(exprs: _*) 

df_noNullsは今いないNULL値を持つ唯一の数値列のデータフレームである必要があり、彼らは列のNULLに置き換えられました。しかし、すべての値（ラベル/ターゲットを引いたもの）のベクトルを作成しようとすると、「アセンブルする値はnullにはなりません」というエラーが出ます。エラーのスクリーンショットを添付しました。また、ユーザー定義関数の実行に失敗したという。

私は最近スカラーについて多くの質問をしてきたことを知っています。申し訳ありませんが、私は本当にこれをやろうとしています。以下は間違いケース内のRFステップへのコードの残りの部分はどこかに存在している。

val num_feat = numCols.filter(! _.contains("call")) 
val features=num_feat 
val featureAssembler = new VectorAssembler().setInputCols(features).setOutputCol("features") 
val reweight_vector = featureAssembler.transform(df_noNulls) 
val rf50 = new RandomForestClassifier().setSeed(9).setLabelCol("call_ind").setFeaturesCol("features").setNumTrees(500).setMaxBins(100).fit(reweight_vector) 

Answer 1

私はこの原因は完全nullである列であることをを推測しています - その場合には、平均値もゼロになります。それを避けるために、あなたは、単にたとえばリテラルの0を使用して、​​3210表現で別の「フォールバック」を追加することができます。

val exprs = numDf.columns.map(c => coalesce(col(c), col(s"avg($c)"), lit(0.0)).alias(c)) 

unchanedコードの残りの部分で、これはdf_noNullsの値のどれを確保するべきではないnullです。