спарк назначить имя столбца для функции withColumn из переменных полей

Question

У меня есть некоторые данные json, как показано ниже, мне нужно создать новые столбцы на основе некоторых значений Jason

{"start": "1234567679", "test": ["abc"], "value": 324," end ":" 1234567689 "}

{" start ":" 1234567679 "," test ": [" xyz "]," value ":" Near "," end ":"1234567689 "}

{" start ":" 1234568679 "," test ": [" pqr "]," value ": [" Attr "," "]," end ":" 1234568679 "}

{"start": "1234568997", "test": ["mno"], "value": ["{\" key \ ": \" 1 \ ", \" value \ ": [\ "789 \"]} "]," end ":" 1234568999 "}

выше - это пример json

Я хочу создать столбец, как показано ниже

 start      abc     xyz    pqr     mno    end
 1234567679 324     null   null    null   1234567689
 1234567889 null    Near   null    null   1234567989
 1234568679 null    null   attr    null   1234568679
 1234568997 null    null   null    789    1234568999

 def getValue1(s1: Seq[String], v: String) = {
      if (s1(0)=="abc"))  v else null
 } 

 def getValue2(s1: Seq[String], v: String) = {
     if (s1(0)=="xyz"))  v else null
 }  

 val df = spark.read.json("path to json")

 val tdf = df.withColumn("abc",getValue1($"test", $"value")).withColumn("xyz",getValue2($"test", $"value"))

Но это я не хочу использовать, потому что мои тестовые значения больше, я хочу, чтобы какая-то функция делала что-то вроде этого

 def getColumnname(s1: Seq[String]) = {
    return s1(0)
 }  


 val tdf = df.withColumn(getColumnname($"test"),$"value"))

это хорошая идея, чтобы изменить значения на столбцы, я хочу так, потому что янеобходимо применить это к некоторому коду машинного обучения, который нуждается в простых столбцах

Answer 1

Вы можете использовать операции поворота, чтобы делать такие вещи.Предполагая, что у вас всегда есть один элемент в вашем массиве для столбца test, вот более простое решение:

import org.apache.spark.sql.functions._
val df = sqlContext.read.json("<yourPath>")
df.withColumn("test", $"test".getItem(0)).groupBy($"start", $"end").pivot("test").agg(first("value")).show
+----------+----------+----+----+
|     start|       end| abc| xyz|
+----------+----------+----+----+
|1234567679|1234567689| 324|null|
|1234567889|1234567689|null| 789|
+----------+----------+----+----+

Если у вас есть несколько значений в столбце test, вы также можете использовать функцию explode;

df.withColumn("test", explode($"test")).groupBy($"start", $"end").pivot("test").agg(first("value")).show

Для получения дополнительной информации:

• Документация Spark
• Блог базы данных
• Средний блог

Надеюсь, это поможет!

Обновление I

На основе ваших комментариев и обновленного вопроса, вот решение, которое вынужно следовать.Я намеренно разделил все операции, так что вы можете легко понять, что вам нужно сделать для дальнейших улучшений;

df.withColumn("value", regexp_replace($"value", "\\[", "")). //1
   withColumn("value", regexp_replace($"value", "\\]", "")). //2
   withColumn("value", split($"value", "\\,")).              //3
   withColumn("test", explode($"test")).                     //4
   withColumn("value", explode($"value")).                   //5
   withColumn("value", regexp_replace($"value", " +", "")).  //6
   filter($"value" !== "").                                  //7
   groupBy($"start", $"end").pivot("test").                  //8
   agg(first("value")).show                                  //9

• Когда выпрочитайте такие файлы json, это даст вам фрейм данных, который имеет столбец value с StringType.Вы не можете напрямую конвертировать StringType в ArrayType, поэтому вам нужно выполнить какие-то трюки, как в строках 1, 2, 3, чтобы конвертировать его в ArrayType. Вы можете выполнять эти операции в одной строке или с помощью одного регулярного выражения или определения udf.Все зависит от вас, Я просто пытаюсь показать вам возможности Apache Spark.

• Теперь у вас есть столбец value с ArrayType.Разнесите этот столбец в строке 5, как мы делали в строке 4 для столбца test.Затем примените операции поворота.