如何将字符串冒号分隔的列转换为MapType?

San*_*mar 5 dataframe apache-spark apache-spark-sql pyspark

我正在尝试将 Dataframe 转换为 RDD,以便将地图(带有键值对)分解为不同的行。

Info = sqlContext.read.format("csv"). \
option("delimiter","\t"). \
option("header", "True"). \
option("inferSchema", "True"). \
load("file.tsv")

DataFrame[ID: int, Date: timestamp, Comments: string]
Run Code Online (Sandbox Code Playgroud)

DF中的样本数据如下。

ID  Date                    Comments    
1   2015-04-30 22:42:49.0   {44:'xxxxxxxx'}
2   2015-05-06 08:53:18.0   {83:'aaaaaaaaa', 175:'bbbbbbbbb', 86:'cccccccccc'}
3   2015-05-13 19:57:13.0   {487:'yyyyyyyyyyy', 48:'zzzzzzzzzzzzzz'}
Run Code Online (Sandbox Code Playgroud)

现在,注释已经在键值对中,但它被读取为字符串,我想将每个键值对分解为不同的行。例如

Expected OUTPUT
ID  Date                     Comments   
1   2015-04-30 22:42:49.0    {44:'xxxxxxxx'}
2   2015-05-06 08:53:18.0    {83:'aaaaaaaaa'}
2   2015-05-06 08:53:18.0    {175:'bbbbbbbbb'}
2   2015-05-06 08:53:18.0    {86:'cccccccccc'}
3   2015-05-13 19:57:13.0    {487:'yyyyyyyyyyy'}
3   2015-05-13 19:57:13.0    {48:'zzzzzzzzzzzzzz'}
Run Code Online (Sandbox Code Playgroud)

我尝试将其转换为 RDD 并应用,flatMap但没有成功。我希望返回所有列。我已经尝试过这个:

Info.rdd.flatMap(lambda x: (x['SearchParams'].split(':'), x))
Run Code Online (Sandbox Code Playgroud)

Sha*_*ica 2

使用 DataFrame API 中提供的splitexplode函数按“,”拆分数据。要创建地图,您需要使用create_map. 此函数需要两个单独的列作为输入。下面是创建两个临时列的示例(再次使用split):

Info.withColumn("Comments", explode(split(col("Comments"), ", ")))
  .withColumn("key", split(col("Comments"), ":").getItem(0))
  .withColumn("value", split(col("Comments"), ":").getItem(1))
  .withColumn("Comments", create_map(col("key"), col("value")))
Run Code Online (Sandbox Code Playgroud)

应该可以像这样缩短它(未经测试):

Info.withColumn("Comments", split(explode(split(col("Comments), ", ")), ":")
  .withColumn("Comments", create_map(col("Comments".getItem(0)), col("Comments").getItem(1)))
Run Code Online (Sandbox Code Playgroud)