python - 在 pyspark 中使用 foreachPartition() 函数时如何知道当前正在运行哪个分区?
问题描述
我需要将分区保存到每个分区具有不同名称的文本文件中。但是在下面的代码片段下运行时,只有一个文件通过覆盖前一个分区来保存。
def chunks(iterator):
chunks.counter += 1
l = (list(iterator))
df = pd.DataFrame(l,index=None)
df.to_csv(parent_path+"C"+str(chunks.counter+1)+".txt", header=None, index=None, sep=' ')
chunks.counter=0
sc.parallelize([1,2,3,4,5,6],num_partions).foreachPartition(chunks)
有什么方法可以知道 pySpark 中当前正在运行哪个分区?
解决方案
def chunks(lst, n):
# Yield successive n-sized chunks from the lst...
for i in (range(0, len(lst), n)):
yield i, lst[i:i + n]
for (index, values) in chunks(range(0, 1e5), 1e3): # change this to int's as per your need otherwise it will give float error or will write range obj itself..
with open(f"{parent_path}_C_{index}.txt", "w") as output:
output.write(str(values)) # converting to str
你甚至可以轻松地将它包装到 joblib 中;)在我看来,我们不需要 PySpark。
推荐阅读
- python - Python:将复杂数组转换为极坐标形式
- javascript - 如何在超时时禁用重新调用 NestJS 控制器?
- javascript - 无效的 csrf 令牌快递
- latex - 如何删除 Latex 中图形和标题之间的空间?
- android - MediaStore.Media.Images Android/ Kotlin:如何向用户隐藏保存在图库中的照片?
- javascript - 将 1 平方公里区域内相同位置的人分组
- c++ - 使用 clang LibASTMatchers 获取文件未找到错误
- ios - productsRequest 从未在应用程序购买中调用 swift 5
- amazon-quicksight - 嵌入式 QuickSight 仪表板,带有指向另一个仪表板的 URL 操作
- angular - Angular 调用 API 或前端句柄