我必须分析几个压缩文件中的8 GB Twitter数据,这是一个月的数据,频率为一小时。每个文件包含几条JSON格式的推文,每行都是推文。
这是我在第一次筛选时使用的代码:
import pandas as pd
import pathlib
import timeit
import zipfile
def main():
twitter_files = list(pathlib.Path("TwitterData").iterdir())
twitter_data = []
for tfilename in twitter_files:
with zipfile.ZipFile(tfilename, "r") as zfile:
# each zip file should contain just one json file,
# this loop it's added for security reason
for filename in zfile.namelist():
print(filename)
with zfile.open(filename) as jfile:
df = pd.read_json(jfile, lines=True)
# Remove NaN and duplicated values
sub_df = df[[
"id_str",
"created_at",
"in_reply_to_status_id_str",
"text",
"entities",
"extended_entities",
"quoted_status_id_str",
"retweeted",
"truncated"
]].dropna(
subset=["id_str", "created_at", "text"],
how="any"
).drop_duplicates(subset="text")
twitter_data.append(sub_df)
del df, sub_df
tweets_df = pd.concat(twitter_data)
del twitter_data
with open("./tweet_data.feather", "w") as ffile:
tweets_df.to_feather(ffile.name)
if __name__ == "__main__":
duration = timeit.timeit(main, number=1)
print(f"time: {duration}")
字符串
我想提取我需要的信息,保存在文件中,然后使用这个文件进行最终分析(找到独特的推文,转推的数量等)。
当然,问题是我的笔记本电脑内存很快就用完了,程序到了一个日期,在月底之前就停止了。
我想找到一个更多的内存和,可能的,时间有效的方式来保存我需要的数据。
我在想逐行读取JSON文件,并保存“好”的推文,但我认为这是一个I/O瓶颈.
你有更好的解决办法吗?
1条答案
按热度按时间rqenqsqc1#
我看到两个选项来改善这一点
*使用较少内存[主内存]:内存不足的原因是内存中一直保留着完整的
tweets_df
(假设你可以读取所有单独的文件,这似乎是事实)。你删除df
和sub_df
和twitter_data
在每个文件被读取/处理后,但你的主框架tweets_df
继续增长无限制。这是不好的。我认为你这样做是因为你直接写.to_feather
,在那里你想要/必须一次写完整的文件内容。解决方案是只关注“清理”工作(阅读单个文件,清理它,并将增量数据保存到某个地方),而不必将所有数据保存在内存中。要做到这一点,通过选择一种可以追加数据的目标格式,以小批量写入数据,而不必始终在内存中保存完整的 Dataframe (最简单的:追加到csv文件,正确的:追加到数据库表,磁盘空间意识妥协:“保存所有数据”的负担应该在磁盘/数据库上,而不是在你的内存变量上。***速度/性能[次要]:**您在阅读每个文件后删除重复文件。您可以在每第2个、第5个、第10个..文件后删除重复文件。您将减少计算工作量,并可能进一步减少内存(您确定您没有跨文件的重复文件吗?)
编辑:第三个改进是并行运行代码(如果实现了步骤1,并且删除了单个主框架
tweets_df
,并且以小批量进行处理,则可能)。然后,您可以在n个单独的示例上运行脚本,每个示例仅处理少量的源文件,可以实现为生成n个单独的输出文件(结构上相等),最终需要协调一次(非常快速的操作),或者动态追加到单个输出文件/数据库中(带有等待条件/写锁)。无论哪种情况,它都是大约1/n的加速(少一点,考虑到写作需要一点等待,或者最后的调解)。