最终结果是OP所期望的一个字典列表,其中所有具有相同“场馆”和“日期”的DataFrame行已经被合并在一起。
# Creating the DataFrames
df_Incoming = sqlContext.createDataFrame([('Hotel','20190101',15,3),('Beach','20190101',93,45)],('Venue','Date','08','10'))
df_Incoming.show()
+
|Venue| Date| 08| 10|
+
|Hotel|20190101| 15| 3|
|Beach|20190101| 93| 45|
+
df_Outgoing = sqlContext.createDataFrame([('Beach','20190101',30,5),('Hotel','20190103',5,15)],('Venue','Date','07','10'))
df_Outgoing.show()
+
|Venue| Date| 07| 10|
+
|Beach|20190101| 30| 5|
|Hotel|20190103| 5| 15|
+
这个想法是从每一行创建一个字典
,并将DataFrame
的所有行
存储为一个大的列表
中的字典。最后一步,我们将那些具有相同Venue
和Date
的字典组合在一起。
由于DataFrame中的所有行
都存储为Row()
对象,因此我们使用collect()函数将所有记录作为Row()
的列表
返回。只是为了说明输出-
print(df_Incoming.collect())
[Row(Venue='Hotel', Date='20190101', 08=15, 10=3), Row(Venue='Beach', Date='20190101', 08=93, 10=45)]
但是,由于我们想要一个包含字典的列表,我们可以使用列表推导式将它们转换为一个 -
list_Incoming = [row.asDict() for row in df_Incoming.collect()]
print(list_Incoming)
[{'10': 3, 'Date': '20190101', 'Venue': 'Hotel', '08': 15}, {'10': 45, 'Date': '20190101', 'Venue': 'Beach', '08': 93}]
但是,由于数字列的形式类似于"08":{ "IncomingCount":15 }
而不是"08":15
,因此我们使用字典推导式
将它们转换为这种形式 -
list_Incoming = [ {k:v if k in ['Venue','Date'] else {'IncomingCount':v} for k,v in dict_element.items()} for dict_element in list_Incoming]
print(list_Incoming)
[{'10': {'IncomingCount': 3}, 'Date': '20190101', 'Venue': 'Hotel', '08': {'IncomingCount': 15}}, {'10': {'IncomingCount': 45}, 'Date': '20190101', 'Venue': 'Beach', '08': {'IncomingCount': 93}}]
同样地,我们也对OutgoingCount
进行操作。
list_Outgoing = [row.asDict() for row in df_Outgoing.collect()]
list_Outgoing = [ {k:v if k in ['Venue','Date'] else {'OutgoingCount':v} for k,v in dict_element.items()} for dict_element in list_Outgoing]
print(list_Outgoing)
[{'10': {'OutgoingCount': 5}, 'Date': '20190101', 'Venue': 'Beach', '07': {'OutgoingCount': 30}}, {'10': {'OutgoingCount': 15}, 'Date': '20190103', 'Venue': 'Hotel', '07': {'OutgoingCount': 5}}]
最后一步:现在,我们已经创建了必要的列表
和字典
,我们需要根据场馆
和日期
将列表合并在一起。
from copy import deepcopy
def merge_lists(list_Incoming, list_Outgoing):
dict1 = {(record['Venue'], record['Date']): record for record in list_Incoming}
result = {}
for record in list_Outgoing:
ckey = record['Venue'], record['Date']
new_record = deepcopy(record)
if ckey in dict1:
for key, value in dict1[ckey].items():
if key in ('Venue', 'Date'):
continue
new_record.setdefault(key, {}).update(value)
result[ckey] = new_record
for key, value in dict1.items():
if key not in result:
result[key] = deepcopy(value)
return list(result.values())
res = merge_lists(list_Incoming, list_Outgoing)
print(res)
[{'10': {'OutgoingCount': 5, 'IncomingCount': 45},
'Date': '20190101',
'Venue': 'Beach',
'08': {'IncomingCount': 93},
'07': {'OutgoingCount': 30}
},
{'10': {'OutgoingCount': 15},
'Date': '20190103',
'Venue': 'Hotel',
'07': {'OutgoingCount': 5}
},
{'10': {'IncomingCount': 3},
'Date': '20190101',
'Venue': 'Hotel',
'08': {'IncomingCount': 15}
}]