如何将包含多个键值对的列拆分为pyspark中的不同列

Question

我正在研究一个非常大的数据集，称为 AWS 上的 Reddit。我首先阅读了一个小样本：

file_lzo = sc.newAPIHadoopFile("s3://mv559/reddit/sample-data/", 
                               "com.hadoop.mapreduce.LzoTextInputFormat", 
                               "org.apache.hadoop.io.LongWritable", 
                               "org.apache.hadoop.io.Text")

所以我得到了一个名为file_lzo 的rdd。我取了第一个元素，数据看起来像：

[(0,
  '{"archived":false,"author":"TistedLogic","author_created_utc":1312615878,"author_flair_background_color":null,"author_flair_css_class":null,"author_flair_richtext":[],"author_flair_template_id":null,"author_flair_text":null,"author_flair_text_color":null,"author_flair_type":"text","author_fullname":"t2_5mk6v","author_patreon_flair":false,"body":"Is it still r\\/BoneAppleTea worthy if it\'s the opposite?","can_gild":true,"can_mod_post":false,"collapsed":false,"collapsed_reason":null,"controversiality":0,"created_utc":1538352000,"distinguished":null,"edited":false,"gilded":0,"gildings":{"gid_1":0,"gid_2":0,"gid_3":0},"id":"e6xucdd","is_submitter":false,"link_id":"t3_9ka1hp","no_follow":true,"parent_id":"t1_e6xu13x","permalink":"\\/r\\/Unexpected\\/comments\\/9ka1hp\\/jesus_fking_woah\\/e6xucdd\\/","removal_reason":null,"retrieved_on":1539714091,"score":2,"send_replies":true,"stickied":false,"subreddit":"Unexpected","subreddit_id":"t5_2w67q","subreddit_name_prefixed":"r\\/Unexpected","subreddit_type":"public"}')]

然后我使用这个 rdd 创建一个数据框

df = spark.createDataFrame(file_lzo,['idx','map_col'])
df.show(4)

看起来像这样

+-----+--------------------+
|  idx|             map_col|
+-----+--------------------+
|    0|{"archived":false...|
|70139|{"archived":false...|
|70139|{"archived":false...|
|70139|{"archived":false...|
+-----+--------------------+
only showing top 4 rows

最后我想以如下所示的数据帧格式获取数据，并将其保存为 S3 中的拼花格式以供将来使用。

我尝试创建一个 Schema，然后使用 read.json，但是我得到的所有值都是 Null

fields = [StructField("archived", BooleanType(), True), 
          StructField("author", StringType(), True),
          StructField("author_flair_css_class", StringType(), True),
          StructField("author_flair_text", StringType(), True),
          StructField("body", StringType(), True),
          StructField("can_gild", BooleanType(), True),         
          StructField("controversiality", LongType(), True),
          StructField("created_utc", StringType(), True),
          StructField("distinguished", StringType(), True),
          StructField("edited", StringType(), True),
          StructField("gilded", LongType(), True), 
          StructField("id", StringType(), True),
          StructField("is_submitter", StringType(), True),
          StructField("link_id", StringType(), True),
          StructField("parent_id", StringType(), True),
          StructField("permalink", StringType(), True),
          StructField("permalink", StringType(), True),
          StructField("removal_reason", StringType(), True),
          StructField("retrieved_on", LongType(), True), 
          StructField("score",LongType() , True),
          StructField("stickied", BooleanType(), True),  
          StructField("subreddit", StringType(), True),
          StructField("subreddit_id", StringType(), True)]

schema = StructType(fields)

+--------+------+----------------------+-----------------+----+--------+----------------+-----------+-------------+------+------+----+------------+-------+---------+---------+---------+--------------+------------+-----+--------+---------+------------+
|archived|author|author_flair_css_class|author_flair_text|body|can_gild|controversiality|created_utc|distinguished|edited|gilded|  id|is_submitter|link_id|parent_id|permalink|permalink|removal_reason|retrieved_on|score|stickied|subreddit|subreddit_id|
+--------+------+----------------------+-----------------+----+--------+----------------+-----------+-------------+------+------+----+------------+-------+---------+---------+---------+--------------+------------+-----+--------+---------+------------+
|    null|  null|                  null|             null|null|    null|            null|       null|         null|  null|  null|null|        null|   null|     null|     null|     null|          null|        null| null|    null|     null|        null|
|    null|  null|                  null|             null|null|    null|            null|       null|         null|  null|  null|null|        null|   null|     null|     null|     null|          null|        null| null|    null|     null|        null|
|    null|  null|                  null|             null|null|    null|            null|       null|         null|  null|  null|null|        null|   null|     null|     null|     null|          null|        null| null|    null|     null|        null|
+--------+------+----------------------+-----------------+----+--------+----------------+-----------+-------------+------+------+----+------------+-------+---------+---------+---------+--------------+------------+-----+--------+---------+------------+

Answer 1

查看您想要的输出，您可以将您的 json 视为 MapType() 的列，然后从中提取列。

开始创建数据框：

my_rdd = [(0, {"author":  "abc", "id": "012", "archived": "False"}),
        (1, {"author": "bcd", "id": "013", "archived": "False"}),
        (2, {"author": "cde", "id": "014", "archived": "True"}),
        (3, {"author": "edf", "id": "015", "archived": "False"})]
df = sqlContext.createDataFrame(my_rdd,['idx','map_col'])
df.show()
# +---+--------------------+
# |idx|             map_col|
# +---+--------------------+
# |  0|Map(id -> 012, au...|
# |  1|Map(id -> 013, au...|
# |  2|Map(id -> 014, au...|
# |  3|Map(id -> 015, au...|
# +---+--------------------+

然后，如果您事先不知道要提取哪些密钥，则收集一个并获取密钥，例如：

from pyspark.sql import functions as f

one = df.select(f.col('map_col')).rdd.take(1)
my_dict = one[0][0].keys()
my_dict
# dict_keys(['id', 'author', 'archived'])

如果您已经知道密钥列表，请直接使用该列表。

因此，您可以将地图列展平：

keep_cols = [f.col('map_col').getItem(k).alias(k) for k in my_dict]
df.select(keep_cols).show()
#+---+------+--------+
#| id|author|archived|
#+---+------+--------+
#|012|   abc|   False|
#|013|   bcd|   False|
#|014|   cde|    True|
#|015|   edf|   False|
#+---+------+--------+

方法getItem() 和alias() 正在发挥作用：第一个从映射列中提取选定的键，第二个根据需要重命名获得的列。