时间戳

首页 > 解决方案 > 根据另一个数据框的行值在数据框中添加新列

scala - 根据另一个数据框的行值在数据框中添加新列

问题描述

我需要向数据框中添加一个新列，DF1但新列的值应使用该DF. 要使用的其他哪些列将在另一个数据框中给出DF2。
例如。DF1 

|protocolNo|serialNum|testMethod  |testProperty|
+----------+---------+------------+------------+       
|Product1  |  AB     |testMethod1 | TP1        |
|Product2  |  CD     |testMethod2 | TP2        |

DF2-

|action| type|               value       |        exploded |
+------------+---------------------------+-----------------+
|append|hash |        [protocolNo]       | protocolNo      |
|append|text |            _              |     _           | 
|append|hash | [serialNum,testProperty]  | serialNum       |
|append|hash | [serialNum,testProperty]  | testProperty    |

现在，如果列类型的值为hash ，则展开列的值DF2将是列名。DF1

Required- 应在 DF1 中创建新列。该值应计算如下 -

hash[protocolNo]_hash[serialNumTestProperty]~~~这里就列的地方他们对应的行值应该来了。

例如。对于 DF1 的第 1 行，列值应为 

hash[Product1]_hash[ABTP1]

这将abc-df_egh-45e在散列后产生类似的结果。

DF1 的每一行都应遵循上述程序。

我尝试在 DF1 上使用 UDF 使用 map 和 withColumn 函数。但是在 UDF 中，外部数据帧值不可访问（给出空指针异常]，我也无法将 DataFrame 作为 UDF 的输入。

如上所述，输入 DF 将是 DF1 和 DF2。

所需输出 DF-

|protocolNo|serialNum|testMethod  |testProperty| newColumn      |
+----------+---------+------------+------------+----------------+       
|Product1  |  AB     |testMethod1 | TP1        | abc-df_egh-4je |
|Product2  |  CD     |testMethod2 | TP2        | dfg-df_ijk-r56 |

newColumn值在散列之后

标签： scalaapache-sparkapache-spark-sql

解决方案

代替 DF2，您可以将 DF2 转换为像 Specifications 这样的案例类，例如

case class Spec(columnName:String,inputColumns:Seq[String],action:String,action:String,type:String*){}

创建上述类的实例

val specifications = Seq(
Spec("new_col_name",Seq("serialNum","testProperty"),"hash","append")
                     )

然后你可以处理下面的列

 val transformed =  specifications
        .foldLeft(dtFrm)((df: DataFrame, spec: Specification) => df.transform(transformColumn(columnSpec)))

def transformColumn(spec: Spec)(df: DataFrame): DataFrame = { 

 spec.type.foldLeft(df)((df: DataFrame, type : String) => {
           type match {
                  case "append" => {have a case match of the action and do that , then append with df.withColumn}

}
}

语法可能不正确

推荐阅读