条件付きで新しい列をSpark（scala）で動的に作成する

Question

プログラムで選択された既存の列から塗りつぶされた新しい条件付き列をsparkで作成しようとしています。

謝罪は、この音は複雑ですが、ここに例があります。サンプルDF：今すぐ

// sample df 
val df = Seq(
    (1, "2014/07/31 23:00:01", 1, 2), 
    (1, "2014/07/30 12:40:32", 3, 3), 
    (1, "2016/08/09 10:12:43", 5, 6)) 
.toDF("id", "date", "7_col", "8_col") 
.withColumn("timestamp", unix_timestamp($"date", "yyyy/MM/dd HH:mm:ss").cast("timestamp")) 

+---+-------------------+-----+-----+-------------------+ 
| id|    date|7_col|8_col|   timestamp| 
+---+-------------------+-----+-----+-------------------+ 
| 1|2014/07/31 23:00:01| 1| 2|2014-07-31 23:00:01| 
| 1|2014/07/30 12:40:32| 3| 3|2014-07-30 12:40:32| 
| 1|2016/08/09 10:12:43| 5| 6|2016-08-09 10:12:43| 
+---+-------------------+-----+-----+-------------------+ 

は、私がtimestamp COLの月が7月であるかどうかに応じて、7_colまたは8_colのいずれかの内容で満たされた新しい列を作成する（7_col）を希望しますまたは第8月（8_col）です。だから、結果はそうのようになります。

+---+-------------------+-----+-----+-------------------+-------+ 
| id|    date|7_col|8_col|   timestamp|new_col| 
+---+-------------------+-----+-----+-------------------+-------+ 
| 1|2014/07/31 23:00:01| 1| 2|2014-07-31 23:00:01|  1| 
| 1|2014/07/30 12:40:32| 3| 3|2014-07-30 12:40:32|  3| 
| 1|2016/08/09 10:12:43| 5| 6|2016-08-09 10:12:43|  6| 
+---+-------------------+-----+-----+-------------------+-------+ 

を今、私は単にIntとして月を渡し、そうのように、渡すために、列の名前の入力にそれを補間する場合、私は部分的にプログラム的にこれを行うことができます：

を私が知っている今

df.withColumn("new_col", $"${month($"timestamp")}_col").show 

Error: org.apache.spark.sql.AnalysisException: cannot resolve '`month(timestamp)_col`' given input columns: [7_col, id, date, 8_col, timestamp]; 

、：代わりに、私はtimestamp列から抽出された月に合格しようとする桁の、それが動作するように失敗したが

df.withColumn("new_col", $"${7}_col").show 

+---+-------------------+-----+-----+-------------------+-------+ 
| id|    date|7_col|8_col|   timestamp|new_col| 
+---+-------------------+-----+-----+-------------------+-------+ 
| 1|2014/07/31 23:00:01| 1| 2|2014-07-31 23:00:01|  1| 
| 1|2014/07/30 12:40:32| 3| 3|2014-07-30 12:40:32|  3| 
| 1|2016/08/09 10:12:43| 5| 6|2016-08-09 10:12:43|  5| 
+---+-------------------+-----+-----+-------------------+-------+ 

、その例えば、私は単純にそうように抽出された月Intと私のnew_colを埋めることができますので、Int結果の月の作品や結果を抽出するコード：

df.withColumn("new_col", month($"timestamp")).show 

+---+-------------------+-----+-----+-------------------+-------+ 
| id|    date|7_col|8_col|   timestamp|new_col| 
+---+-------------------+-----+-----+-------------------+-------+ 
| 1|2014/07/31 23:00:01| 1| 2|2014-07-31 23:00:01|  7| 
| 1|2014/07/30 12:40:32| 3| 3|2014-07-30 12:40:32|  7| 
| 1|2016/08/09 10:12:43| 5| 6|2016-08-09 10:12:43|  8| 
+---+-------------------+-----+-----+-------------------+-------+ 

をしかし、私は、私は、このような渡すことができない理由を把握することはできませんIntを作成し、列名に補間します。

アイデア？

Answer 1

あなたはwhen.otherwiseを使用することができます。

df.withColumn("new_col", when(month($"timestamp") === 7, $"7_col").otherwise($"8_col")).show 
+---+-------------------+-----+-----+-------------------+-------+ 
| id|    date|7_col|8_col|   timestamp|new_col| 
+---+-------------------+-----+-----+-------------------+-------+ 
| 1|2014/07/31 23:00:01| 1| 2|2014-07-31 23:00:01|  1| 
| 1|2014/07/30 12:40:32| 3| 3|2014-07-30 12:40:32|  3| 
| 1|2016/08/09 10:12:43| 5| 6|2016-08-09 10:12:43|  6| 
+---+-------------------+-----+-----+-------------------+-------+ 

---

動的month_colを処理するための別のオプションを：

val months = (7 to 8).map(m => when(month(col("timestamp")) === m, col(s"${m}_col"))) 
//  change 7 to 8 to a sequence of all exsiting months columns 

df.withColumn("new_col", coalesce(months: _*)).show 
+---+-------------------+-----+-----+-------------------+-------+ 
| id|    date|7_col|8_col|   timestamp|new_col| 
+---+-------------------+-----+-----+-------------------+-------+ 
| 1|2014/07/31 23:00:01| 1| 2|2014-07-31 23:00:01|  1| 
| 1|2014/07/30 12:40:32| 3| 3|2014-07-30 12:40:32|  3| 
| 1|2016/08/09 10:12:43| 5| 6|2016-08-09 10:12:43|  6| 
+---+-------------------+-----+-----+-------------------+-------+