如何使用 Spark 2 屏蔽列？答案

【问题标题】：How to mask columns using Spark 2?如何使用 Spark 2 屏蔽列？
【发布时间】：2018-03-02 02:04:11
【问题描述】：

我有一些表需要屏蔽其中的一些列。要屏蔽的列因表而异，我正在从application.conf 文件中读取这些列。

例如下图的员工表

+----+------+-----+---------+
| id | name | age | address |
+----+------+-----+---------+
| 1  | abcd | 21  | India   |
+----+------+-----+---------+
| 2  | qazx | 42  | Germany |
+----+------+-----+---------+

如果我们想屏蔽姓名和年龄列，那么我会按顺序获取这些列。

val mask = Seq("name", "age")

屏蔽后的预期值为：

+----+----------------+----------------+---------+
| id | name           | age            | address |
+----+----------------+----------------+---------+
| 1  | *** Masked *** | *** Masked *** | India   |
+----+----------------+----------------+---------+
| 2  | *** Masked *** | *** Masked *** | Germany |
+----+----------------+----------------+---------+

如果我有员工表一个数据框，那么屏蔽这些列的方法是什么？

如果我有如下所示的payment 表并且想要屏蔽name 和salary 列，那么我将序列中的屏蔽列设为

+----+------+--------+----------+
| id | name | salary | tax_code |
+----+------+--------+----------+
| 1  | abcd | 12345  | KT10     |
+----+------+--------+----------+
| 2  | qazx | 98765  | AD12d    |
+----+------+--------+----------+

val mask = Seq("name", "salary")

我尝试了类似 mask.foreach(c => base.withColumn(c, regexp_replace(col(c), "^.*?$", "*** Masked ***" ) ) ) 的操作，但没有返回任何内容。

感谢@philantrovert，我找到了解决方案。这是我使用的解决方案：

def maskData(base: DataFrame, maskColumns: Seq[String]) = {
    val maskExpr = base.columns.map { col => if(maskColumns.contains(col)) s"'*** Masked ***' as ${col}" else col }
    base.selectExpr(maskExpr: _*)
}

【问题讨论】：

标签： scala apache-spark apache-spark-sql apache-spark-2.0

【解决方案1】：

最简单、最快的方法是使用withColumn 并简单地用"*** Masked ***" 覆盖列中的值。使用您的小型示例数据框

val df = spark.sparkContext.parallelize( Seq (
  (1, "abcd", 12345, "KT10" ),
  (2, "qazx", 98765, "AD12d")
)).toDF("id", "name", "salary", "tax_code")

如果您有少量的列要被屏蔽，并且名称已知，那么您可以简单地这样做：

val mask = Seq("name", "salary")

df.withColumn("name", lit("*** Masked ***"))
  .withColumn("salary", lit("*** Masked ***"))

否则，你需要创建一个循环：

var df2 = df
for (col <- mask){
  df2 = df2.withColumn(col, lit("*** Masked ***"))
}

这两种方法都会给你这样的结果：

+---+--------------+--------------+--------+
| id|          name|        salary|tax_code|
+---+--------------+--------------+--------+
|  1|*** Masked ***|*** Masked ***|    KT10|
|  2|*** Masked ***|*** Masked ***|   AD12d|
+---+--------------+--------------+--------+

【讨论】：

【解决方案2】：

请检查下面的代码。关键是udf 函数。

val df = ss.sparkContext.parallelize( Seq (
  ("c1", "JAN-2017", 49 ),
  ("c1", "MAR-2017", 83),
)).toDF("city", "month", "sales")
df.show()

val mask = udf( (s : String) => {
  "*** Masked ***"
})

df.withColumn("city", mask($"city")).show`

【讨论】：

【解决方案3】：

你的陈述

mask.foreach(c => base.withColumn(c, regexp_replace(col(c), "^.*?$", "*** Masked ***" ) ) )

将返回一个听起来不太好的List[org.apache.spark.sql.DataFrame]。

您可以使用selectExpr 并使用以下方法生成您的regexp_replace 表达式：

base.show
+---+----+-----+-------+
| id|name|  age|address|
+---+----+-----+-------+
|  1|abcd|12345|  KT10 |
|  2|qazx|98765|  AD12d|
+---+----+-----+-------+

val mask = Seq("name", "age")
val expr = df.columns.map { col =>
   if (mask.contains(col) ) s"""regexp_replace(${col}, "^.*", "** Masked **" ) as ${col}"""
   else col
 }

这将为序列mask中存在的列生成一个带有 regex_replace 的表达式

Array[String] = Array(id, regexp_replace(name, "^.*", "** Masked **" ) as name, regexp_replace(age, "^.*", "** Masked **" ) as age, address)

现在您可以在生成的序列上使用selectExpr

base.selectExpr(expr: _*).show

+---+------------+------------+-------+
| id|        name|         age|address|
+---+------------+------------+-------+
|  1|** Masked **|** Masked **|  KT10 |
|  2|** Masked **|** Masked **|  AD12d|
+---+------------+------------+-------+

【讨论】：