星火 - 从行数据框中删除特殊字符的不同列类型

Question

假设我有一个数据帧包含多列，有些类型串其他类型INT和其他类型地图。

例如 场/列types: stringType|intType|mapType<string,int>|...

|-------------------------------------------------------------------------- 
| myString1  |myInt1| myMap1            |... 
|-------------------------------------------------------------------------- 
|"this_is_#string"| 123 |{"str11_in#map":1,"str21_in#map":2, "str31_in#map": 31}|... 
|"this_is_#string"| 456 |{"str12_in#map":1,"str22_in#map":2, "str32_in#map": 32}|... 
|"this_is_#string"| 789 |{"str13_in#map":1,"str23_in#map":2, "str33_in#map": 33}|... 
|-------------------------------------------------------------------------- 

我想删除像“_”和字符串和地图类型的所有列“＃” 所以结果据帧/ RDD会出现一些字符：

|------------------------------------------------------------------------ 
|myString1  |myInt1|  myMap1|...         | 
|------------------------------------------------------------------------ 
|"thisisstring"| 123 |{"str11inmap":1,"str21inmap":2, "str31inmap": 31}|... 
|"thisisstring"| 456 |{"str12inmap":1,"str22inmap":2, "str32inmap": 32}|... 
|"thisisstring"| 789 |{"str13inmap":1,"str23inmap":2, "str33inmap": 33}|... 
|------------------------------------------------------------------------- 

我不确定是否最好将Dataframe转换为RDD并使用它或在Dataframe中执行工作。

此外，不知道如何以最佳方式处理不同列类型的正则表达式（我唱歌斯卡拉）。 我想执行此操作，这两种类型（字符串和地图）的所有列，尽量避免使用像列名：

def cleanRows(mytabledata: DataFrame): RDD[String] = { 

//this will do the work for a specific column (myString1) of type string 
val oneColumn_clean = mytabledata.withColumn("myString1", regexp_replace(col("myString1"),"[_#]","")) 

     ... 
//return type can be RDD or Dataframe... 
} 

有没有简单的解决方案来执行呢？ 由于

Answer 1

一种选择是，定义两个UDF的处理字符串类型列和分别地图类型柱：

import org.apache.spark.sql.functions.udf 
val df = Seq(("this_is#string", 3, Map("str1_in#map" -> 3))).toDF("myString", "myInt", "myMap") 
df.show 
+--------------+-----+--------------------+ 
|  myString|myInt|    myMap| 
+--------------+-----+--------------------+ 
|this_is#string| 3|Map(str1_in#map -...| 
+--------------+-----+--------------------+ 

1）UDF处理字符串型柱：

def remove_string: String => String = _.replaceAll("[_#]", "") 
def remove_string_udf = udf(remove_string) 

2）UDF以处理地图类型列：

def remove_map: Map[String, Int] => Map[String, Int] = _.map{ case (k, v) => k.replaceAll("[_#]", "") -> v } 
def remove_map_udf = udf(remove_map) 

3）将udfs应用到相应的公司列清理它：

df.withColumn("myString", remove_string_udf($"myString")). 
    withColumn("myMap", remove_map_udf($"myMap")).show 

+------------+-----+-------------------+ 
| myString|myInt|    myMap| 
+------------+-----+-------------------+ 
|thisisstring| 3|Map(str1inmap -> 3)| 
+------------+-----+-------------------+