空の文字列をDataFrameのNone/Null値に置き換えます

Question

私はnullと空の文字列を同じ列に入れたSpark 1.5.0 DataFrameを持っています。私はすべての列のすべての空の文字列をnull（None、Python）に変換したいと思います。 DataFrameには何百もの列があるので、各列のハードコード化された操作を避けようとしています。

以下の試行を参照してください。エラーが発生します。

from pyspark.sql import SQLContext 
sqlContext = SQLContext(sc) 

## Create a test DataFrame 
testDF = sqlContext.createDataFrame([Row(col1='foo', col2=1), Row(col1='', col2=2), Row(col1=None, col2='')]) 
testDF.show() 
## +----+----+ 
## |col1|col2| 
## +----+----+ 
## | foo| 1| 
## | | 2| 
## |null|null| 
## +----+----+ 

## Try to replace an empty string with None/null 
testDF.replace('', None).show() 
## ValueError: value should be a float, int, long, string, list, or tuple 

## A string value of null (obviously) doesn't work... 
testDF.replace('', 'null').na.drop(subset='col1').show() 
## +----+----+ 
## |col1|col2| 
## +----+----+ 
## | foo| 1| 
## |null| 2| 
## +----+----+ 

Answer 1

それはこのように簡単です：

from pyspark.sql.functions import col, when 

def blank_as_null(x): 
    return when(col(x) != "", col(x)).otherwise(None) 

dfWithEmptyReplaced = testDF.withColumn("col1", blank_as_null("col1")) 

dfWithEmptyReplaced.show() 
## +----+----+ 
## |col1|col2| 
## +----+----+ 
## | foo| 1| 
## |null| 2| 
## |null|null| 
## +----+----+ 

dfWithEmptyReplaced.na.drop().show() 
## +----+----+ 
## |col1|col2| 
## +----+----+ 
## | foo| 1| 
## +----+----+ 

あなたが複数の列を埋めるためにしたい場合は、たとえば減らすことができます。

to_convert = set([...]) # Some set of columns 

reduce(lambda df, x: df.withColumn(x, blank_as_null(x)), to_convert, testDF) 

や理解を使用する：

exprs = [ 
    blank_as_null(x).alias(x) if x in to_convert else x for x in testDF.columns] 

testDF.select(*exprs) 

具体的には文字列フィールドのレートは、でthe answerを確認してください。

Answer 2

私のソリューションは、あなたが望む限り多くの分野に対処することができI'vは、これまで見て、すべてのソリューションよりもはるかに優れている、次のように少しの関数を参照してください。

// Replace empty Strings with null values 
    private def setEmptyToNull(df: DataFrame): DataFrame = { 
    val exprs = df.schema.map { f => 
     f.dataType match { 
     case StringType => when(length(col(f.name)) === 0, lit(null: String).cast(StringType)).otherwise(col(f.name)).as(f.name) 
     case _ => col(f.name) 
     } 
    } 

    df.select(exprs: _*) 
    } 

あなたは簡単に書き換えることができます上記のPythonの関数

私は単にzero323さんとsoulmachineの答えの上に追加@liancheng

Answer 3

からこのトリックを学びました。すべてのStringTypeフィールドを変換します。

from pyspark.sql.types import StringType 
string_fields = [] 
for i, f in enumerate(test_df.schema.fields): 
    if isinstance(f.dataType, StringType): 
     string_fields.append(f.name)