別の列の値に基づいてデータフレームの列を更新する

Question

Scalaの別の列の値を使用して列の値を更新しようとしています。

これは私のデータフレーム内のデータです：

ここ

+-------------------+------+------+-----+------+----+--------------------+-----------+ 
|UniqueRowIdentifier| _c0| _c1| _c2| _c3| _c4|     _c5|isBadRecord| 
+-------------------+------+------+-----+------+----+--------------------+-----------+ 
|     1|  0|  0| Name|  0|Desc|     |   0| 
|     2| 2.11| 10000|Juice|  0| XYZ|2016/12/31 : Inco...|   0| 
|     3|-0.500|-24.12|Fruit| -255| ABC| 1994-11-21 00:00:00|   0| 
|     4| 0.087| 1222|Bread|-22.06| | 2017-02-14 00:00:00|   0| 
|     5| 0.087| 1222|Bread|-22.06| |     |   0| 
+-------------------+------+------+-----+------+----+--------------------+-----------+ 

列_c5が間違っていた値が含まれています（行2の値は、文字列誤ったを持っている）私は更新したいそれに基づいてisBadRecordフィールドを1に設定します。

このフィールドを更新する方法はありますか？

Answer 1

withColumn apiを使用して、functionsのいずれかを使用して、不適切なレコードとして1を入力してください。あなたのケースでは

あなたは

def fillbad = udf((c5 : String) => if(c5.contains("Incorrect")) 1 else 0) 

udf機能を書いて、最良のオプションは、UDFを作成し、それが日付フォーマットを行う変換しようとすることです

val newDF = dataframe.withColumn("isBadRecord", fillbad(dataframe("_c5"))) 

Answer 2

としてそれを呼び出すことができます。その後、0を返す変換できる場合、あなたが悪いの日付形式にこのことができます

 val spark = SparkSession.builder().master("local") 
     .appName("test").getOrCreate() 

     import spark.implicits._ 

//create test dataframe 
     val data = spark.sparkContext.parallelize(Seq(
     (1,"1994-11-21 Xyz"), 
     (2,"1994-11-21 00:00:00"), 
     (3,"1994-11-21 00:00:00") 
    )).toDF("id", "date") 

// create udf which tries to convert to date format 
// returns 0 if success and returns 1 if failure 
     val check = udf((value: String) => { 
     Try(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(value)) match { 
      case Success(d) => 1 
      case Failure(e) => 0 
     } 
     }) 

// Add column 
     data.withColumn("badData", check($"date")).show 

希望を持っている場合でも 他に1つの

この作品を返します！

Answer 3

SQLを更新することについて推論するのではなく、SQLの場合と同様に考えることをお勧めします。次の操作を行うことができます：ここで

import org.spark.sql.functions.when 

val spark: SparkSession = ??? // your spark session 
val df: DataFrame = ??? // your dataframe 

import spark.implicits._ 

df.select(
    $"UniqueRowIdentifier", $"_c0", $"_c1", $"_c2", $"_c3", $"_c4", 
    $"_c5", when($"_c5".contains("Incorrect"), 1).otherwise(0) as "isBadRecord") 

は、ローカルで結果を見るためにあなたのスパークシェル上でコピーして貼り付けることができます自己完結型のスクリプトです：

誰の関連出力以下れる

import org.apache.spark.sql.Row 
import org.apache.spark.sql.types._ 

sc.setLogLevel("ERROR") 

val schema = 
    StructType(Seq(
    StructField("UniqueRowIdentifier", IntegerType), 
    StructField("_c0", DoubleType), 
    StructField("_c1", DoubleType), 
    StructField("_c2", StringType), 
    StructField("_c3", DoubleType), 
    StructField("_c4", StringType), 
    StructField("_c5", StringType), 
    StructField("isBadRecord", IntegerType))) 

val contents = 
    Seq(
    Row(1, 0.0 ,  0.0 , "Name", 0.0, "Desc",      "", 0), 
    Row(2, 2.11 , 10000.0 , "Juice", 0.0, "XYZ", "2016/12/31 : Incorrect", 0), 
    Row(3, -0.5 , -24.12, "Fruit", -255.0, "ABC", "1994-11-21 00:00:00", 0), 
    Row(4, 0.087, 1222.0 , "Bread", -22.06, "", "2017-02-14 00:00:00", 0), 
    Row(5, 0.087, 1222.0 , "Bread", -22.06, "",      "", 0) 
) 

val df = spark.createDataFrame(sc.parallelize(contents), schema) 

df.show() 

val withBadRecords = 
    df.select(
    $"UniqueRowIdentifier", $"_c0", $"_c1", $"_c2", $"_c3", $"_c4", 
    $"_c5", when($"_c5".contains("Incorrect"), 1).otherwise(0) as "isBadRecord") 

withBadRecords.show() 

：

+-------------------+-----+-------+-----+------+----+--------------------+-----------+ 
|UniqueRowIdentifier| _c0| _c1| _c2| _c3| _c4|     _c5|isBadRecord| 
+-------------------+-----+-------+-----+------+----+--------------------+-----------+ 
|     1| 0.0| 0.0| Name| 0.0|Desc|     |   0| 
|     2| 2.11|10000.0|Juice| 0.0| XYZ|2016/12/31 : Inco...|   0| 
|     3| -0.5| -24.12|Fruit|-255.0| ABC| 1994-11-21 00:00:00|   0| 
|     4|0.087| 1222.0|Bread|-22.06| | 2017-02-14 00:00:00|   0| 
|     5|0.087| 1222.0|Bread|-22.06| |     |   0| 
+-------------------+-----+-------+-----+------+----+--------------------+-----------+ 

+-------------------+-----+-------+-----+------+----+--------------------+-----------+ 
|UniqueRowIdentifier| _c0| _c1| _c2| _c3| _c4|     _c5|isBadRecord| 
+-------------------+-----+-------+-----+------+----+--------------------+-----------+ 
|     1| 0.0| 0.0| Name| 0.0|Desc|     |   0| 
|     2| 2.11|10000.0|Juice| 0.0| XYZ|2016/12/31 : Inco...|   1| 
|     3| -0.5| -24.12|Fruit|-255.0| ABC| 1994-11-21 00:00:00|   0| 
|     4|0.087| 1222.0|Bread|-22.06| | 2017-02-14 00:00:00|   0| 
|     5|0.087| 1222.0|Bread|-22.06| |     |   0| 
+-------------------+-----+-------+-----+------+----+--------------------+-----------+