Dask DataFrame.to_parquet 读取失败 - 重新分区 - 写入操作
Dask DataFrame.to_parquet fails on read - repartition - write operation
我有以下工作流程。
def read_file(path, indx):
df = pd.read_parquet(path)
df.index = [indx] * len(df)
return df
files_list = get_all_files() # list of 10k parquet files, each about 1MB
df = dask.dataframe.from_delayed([dask.delayed(read_file)(x, indx) for (indx, x) in enumerate(files_list)])
df.divisions = list(range(10000)) + [9999] # each divisions include 1 file
new_divisions = [0, 10, 23, 45, ...., 9999] # new_divisions that reduces number of partitions by putting a bunch of files into same partitions.
df = df.repartition(divisions = new_divisions)
df.to_parquet("fewer_files") # This causes dask to essentially freeze and no files get written
选择新的分区，使每个分区中文件的总内存不超过1000 MB。但是，最后的 to_parquet 调用永远挂起。在 dask 仪表板上，没有 activity。所有工作人员消耗的内存仍然很小（55MB），至少在仪表板中是这样；但我怀疑它可能只是没有更新，因为一切都变得非常慢。 python 进程 运行 代码不断增加内存消耗（Mac 中的虚拟内存不断增加；我让它增加到 30GB）。
如果 files_list 中只有大约 200 个文件，代码就可以正常工作。这是 df.visualize() 在 files_list 中有 236 个文件被重新分区为 41 个分区时的样子：
知道在有 10k 个文件时可能导致 df.to_parquet 冻结的原因吗？当我在计算前打印 df 时，它显示以下内容：
npartitions=65, Dask Name: repartition-merge, 26417 tasks
此外，我可以让 df.get_partition(0).to_parquet 或其他分区快速工作。但是，整个数据集上的 df.to_parquet 失败。对于我的笔记本电脑中的 4 个工作人员来说，26K 的任务是否太多了？
The new divisions are chosen so that the total memory of the files in each partition doesn't exceed 1000 MB.
如果重新分区的主要考虑因素是内存，那么使用 .repartition(partition_size='1000MB') 可能是个好主意。该脚本如下所示：
df = dask.dataframe.from_delayed([dask.delayed(read_file)(x, indx) for (indx, x) in enumerate(files_list)])
df = df.repartition(partition_size='1000MB')
df.to_parquet("fewer_files")
当 Dask 读取 Parquet 文件时，它会打开文件并可能尝试读取内存中的一个分区以推断文件包含的数据类型。这取决于所使用的引擎（pyarrow 或 fastparquet）以及是否存在 _metadata 文件）。打开文件和推断内存类型都是缓慢的操作。
我怀疑这就是为什么您可以 运行 在 200 个文件上进行计算，而在 10k 个文件上却停滞不前的原因。
一个好的做法是在编写数据时考虑您希望如何读取数据。
对于您的具体情况，我建议一次读取 200 个文件的 parquet 数据，然后在内存中写入更少数量的 parquet 文件。通常，文件数量等于您可用的工作人员数量是明智的，假设来自单个文件的数据适合工作人员内存。
向您的 OS 询问 10.000 个文件句柄也可能导致一些意外行为，这也可能是一个原因。
运行 我的 mac 系统上的这个命令告诉我我的 OS 默认限制我打开 256 个文件描述符：
launchctl limit maxfiles
maxfiles 256 unlimited
尽可能使用 dask.dataframe.read_parquet 或其他 dask I/O 实现，而不是 dask.delayed 包装 pandas I/O 操作。让 dask 直接访问文件对象或文件路径允许调度程序快速评估作业中的步骤并准确估计作业大小和要求，而无需执行完整的工作流程。
说明
通过将 dask.delayed 与 pandas read_parquet reader 一起使用，您实际上是在剥夺 dask 窥视文件结构以帮助安排时间的能力作业，并在 运行 完整作业时多次打开和关闭文件（您甚至还没有解决这个问题）。
当一切都整齐地放入内存时，使用 dask.dataframe.read_parquet 和您使用的延迟方法非常相似。当最佳策略不是简单地“读取所有数据然后弄清楚如何处理它”时，就会出现差异。具体来说，您正在执行许多重建索引和排序操作，所有这些操作都需要 dask 在甚至可以安排 index-manipulation 操作之前了解很多关于文件内容的信息。
本质上，在 dask.delayed 中包装一些东西告诉 dask“这是这个未知的代码块。运行 它作为一个 pure-python 黑盒子很多次。dask.dataframe和 dask.array 接口相比它们的 pandas 和 numpy 接口具有更小的 API 和更少的互操作性，但是你得到的是 dask 实际上知道引擎盖下发生了什么并且可以为你优化它。当你使用 dask.delayed，您将获得灵活性 ，但代价是 dask 为您调整操作的能力。
例子
作为例子，我将创建大量的小文件：
In [9]: tinydf = pd.DataFrame({"col1": [11, 21], "col2": [12, 22]})
...: for i in range(1000):
...: tinydf.to_parquet(f"myfile_{i}.parquet")
dask.dataframe.read_parquet
现在，让我们用 dask.dataframe.read_parquet 阅读这个：
In [10]: df = dask.dataframe.read_parquet([f"myfile_{i}.parquet" for i in range(1000)])
请注意，这快如闪电。我们可以通过检查 dask 属性来查看 high-level 任务图：
In [13]: df.dask
Out[13]:
HighLevelGraph with 1 layers.
<dask.highlevelgraph.HighLevelGraph object at 0x15f79e2f0>
0. read-parquet-e38709bfe39c7f8dfb5c4abf2fd08b50
请注意，dask.dataframe.read_parquet 是一个简单的概念。它可以在此任务中根据需要进行调整和优化。这包括在不读取所有数据的情况下“查看”文件以了解其列结构、查看元数据 file/attributes 等。
In [30]: df.divisions = list(range(0, 2001, 2))
In [31]: df = df.repartition(divisions=list(range(0, 2001, 500)))
In [33]: df.dask
Out[33]:
HighLevelGraph with 2 layers.
<dask.highlevelgraph.HighLevelGraph object at 0x168b5fcd0>
0. read-parquet-e38709bfe39c7f8dfb5c4abf2fd08b50
1. repartition-merge-bc42fb2f09234f7656901995bf3b29fa
完整工作流程的高级图表有两个步骤！ Dask 了解文件 I/O 和重新分区方面的操作。它可以决定如何拆分这些任务，以保持在内存限制内并在工作人员之间分配工作负载，所有这些都不会拖慢调度程序。
dask.delayed(pd.read_parquet)
另一方面，如果我们用 dask.delayed 这样做会怎样？
In [14]: def read_file(path, indx):
...: df = pd.read_parquet(path)
...: df.index = [indx] * len(df)
...: return df
...:
...:
...: files_list = [f"myfile_{i}.parquet" for i in range(1000)]
...: df = dask.dataframe.from_delayed(
...: [dask.delayed(read_file)(x, indx) for (indx, x) in enumerate(files_list)]
...: )
dataframe 预览最终看起来很相似，但如果我们深入查看高级任务图，我们会发现 dask 需要读入所有数据才能知道索引是什么样子！
In [16]: df.dask
Out[16]:
HighLevelGraph with 1001 layers.
<dask.highlevelgraph.HighLevelGraph object at 0x168bf6230>
0. read_file-b7aed020-1dc7-4872-a37d-d514b407a7d8
1. read_file-a0462606-999b-4af1-9977-acb562edab67
2. read_file-286df439-df34-4a5a-baf9-75dd0a5ae09b
3. read_file-4db8c178-a67e-4775-b117-228ac607f02f
4. read_file-a19d6144-5560-4da7-a1f5-8dc92b3ccf1c
# yeah... really there are 1000 of these...
998. read_file-d0cbd4a4-c255-4a77-a905-199bc289a0b5
999. read_file-45a80080-426a-48fd-8dcb-9ba7565307f1
1000. from-delayed-833eff6e232da1e10ca7221b961c21c1
更糟糕的是，每个 pd.read_parquet 使用默认的 pandas 读取行为，即假设数据可以装入内存并立即读取整个文件。 Pandas 不是 return 文件对象 - 它加载所有数据并 return 在 dask 甚至看到它之前是一个 DataFrame。
正因为如此，dask 基本上无法进入调度位，直到所有读取都已完成，并且在工作负载平衡、内存管理等方面几乎没有用处。它可以尝试通过执行第一个任务来获得 sneak-peek 的工作负载，但这仍然是对整个第一个文件的读取。
当我们开始尝试重新排列索引时，情况只会变得更糟。我不会在这里深入，但你明白了......
我有以下工作流程。
def read_file(path, indx):
df = pd.read_parquet(path)
df.index = [indx] * len(df)
return df
files_list = get_all_files() # list of 10k parquet files, each about 1MB
df = dask.dataframe.from_delayed([dask.delayed(read_file)(x, indx) for (indx, x) in enumerate(files_list)])
df.divisions = list(range(10000)) + [9999] # each divisions include 1 file
new_divisions = [0, 10, 23, 45, ...., 9999] # new_divisions that reduces number of partitions by putting a bunch of files into same partitions.
df = df.repartition(divisions = new_divisions)
df.to_parquet("fewer_files") # This causes dask to essentially freeze and no files get written
选择新的分区，使每个分区中文件的总内存不超过1000 MB。但是，最后的 to_parquet 调用永远挂起。在 dask 仪表板上，没有 activity。所有工作人员消耗的内存仍然很小（55MB），至少在仪表板中是这样；但我怀疑它可能只是没有更新，因为一切都变得非常慢。 python 进程 运行 代码不断增加内存消耗（Mac 中的虚拟内存不断增加；我让它增加到 30GB）。
如果 files_list 中只有大约 200 个文件，代码就可以正常工作。这是 df.visualize() 在 files_list 中有 236 个文件被重新分区为 41 个分区时的样子：
知道在有 10k 个文件时可能导致 df.to_parquet 冻结的原因吗？当我在计算前打印 df 时，它显示以下内容：
npartitions=65, Dask Name: repartition-merge, 26417 tasks
此外，我可以让 df.get_partition(0).to_parquet 或其他分区快速工作。但是，整个数据集上的 df.to_parquet 失败。对于我的笔记本电脑中的 4 个工作人员来说，26K 的任务是否太多了？
The new divisions are chosen so that the total memory of the files in each partition doesn't exceed 1000 MB.
如果重新分区的主要考虑因素是内存，那么使用 .repartition(partition_size='1000MB') 可能是个好主意。该脚本如下所示：
df = dask.dataframe.from_delayed([dask.delayed(read_file)(x, indx) for (indx, x) in enumerate(files_list)])
df = df.repartition(partition_size='1000MB')
df.to_parquet("fewer_files")
当 Dask 读取 Parquet 文件时，它会打开文件并可能尝试读取内存中的一个分区以推断文件包含的数据类型。这取决于所使用的引擎（pyarrow 或 fastparquet）以及是否存在 _metadata 文件）。打开文件和推断内存类型都是缓慢的操作。
我怀疑这就是为什么您可以 运行 在 200 个文件上进行计算，而在 10k 个文件上却停滞不前的原因。
一个好的做法是在编写数据时考虑您希望如何读取数据。
对于您的具体情况，我建议一次读取 200 个文件的 parquet 数据，然后在内存中写入更少数量的 parquet 文件。通常，文件数量等于您可用的工作人员数量是明智的，假设来自单个文件的数据适合工作人员内存。
向您的 OS 询问 10.000 个文件句柄也可能导致一些意外行为，这也可能是一个原因。
运行 我的 mac 系统上的这个命令告诉我我的 OS 默认限制我打开 256 个文件描述符：
launchctl limit maxfiles
maxfiles 256 unlimited
尽可能使用 dask.dataframe.read_parquet 或其他 dask I/O 实现，而不是 dask.delayed 包装 pandas I/O 操作。让 dask 直接访问文件对象或文件路径允许调度程序快速评估作业中的步骤并准确估计作业大小和要求，而无需执行完整的工作流程。
说明
通过将 dask.delayed 与 pandas read_parquet reader 一起使用，您实际上是在剥夺 dask 窥视文件结构以帮助安排时间的能力作业，并在 运行 完整作业时多次打开和关闭文件（您甚至还没有解决这个问题）。
当一切都整齐地放入内存时，使用 dask.dataframe.read_parquet 和您使用的延迟方法非常相似。当最佳策略不是简单地“读取所有数据然后弄清楚如何处理它”时，就会出现差异。具体来说，您正在执行许多重建索引和排序操作，所有这些操作都需要 dask 在甚至可以安排 index-manipulation 操作之前了解很多关于文件内容的信息。
本质上，在 dask.delayed 中包装一些东西告诉 dask“这是这个未知的代码块。运行 它作为一个 pure-python 黑盒子很多次。dask.dataframe和 dask.array 接口相比它们的 pandas 和 numpy 接口具有更小的 API 和更少的互操作性，但是你得到的是 dask 实际上知道引擎盖下发生了什么并且可以为你优化它。当你使用 dask.delayed，您将获得灵活性 ，但代价是 dask 为您调整操作的能力。
例子
作为例子，我将创建大量的小文件：
In [9]: tinydf = pd.DataFrame({"col1": [11, 21], "col2": [12, 22]})
...: for i in range(1000):
...: tinydf.to_parquet(f"myfile_{i}.parquet")
dask.dataframe.read_parquet
现在，让我们用 dask.dataframe.read_parquet 阅读这个：
In [10]: df = dask.dataframe.read_parquet([f"myfile_{i}.parquet" for i in range(1000)])
请注意，这快如闪电。我们可以通过检查 dask 属性来查看 high-level 任务图：
In [13]: df.dask
Out[13]:
HighLevelGraph with 1 layers.
<dask.highlevelgraph.HighLevelGraph object at 0x15f79e2f0>
0. read-parquet-e38709bfe39c7f8dfb5c4abf2fd08b50
请注意，dask.dataframe.read_parquet 是一个简单的概念。它可以在此任务中根据需要进行调整和优化。这包括在不读取所有数据的情况下“查看”文件以了解其列结构、查看元数据 file/attributes 等。
In [30]: df.divisions = list(range(0, 2001, 2))
In [31]: df = df.repartition(divisions=list(range(0, 2001, 500)))
In [33]: df.dask
Out[33]:
HighLevelGraph with 2 layers.
<dask.highlevelgraph.HighLevelGraph object at 0x168b5fcd0>
0. read-parquet-e38709bfe39c7f8dfb5c4abf2fd08b50
1. repartition-merge-bc42fb2f09234f7656901995bf3b29fa
完整工作流程的高级图表有两个步骤！ Dask 了解文件 I/O 和重新分区方面的操作。它可以决定如何拆分这些任务，以保持在内存限制内并在工作人员之间分配工作负载，所有这些都不会拖慢调度程序。
dask.delayed(pd.read_parquet)
另一方面，如果我们用 dask.delayed 这样做会怎样？
In [14]: def read_file(path, indx):
...: df = pd.read_parquet(path)
...: df.index = [indx] * len(df)
...: return df
...:
...:
...: files_list = [f"myfile_{i}.parquet" for i in range(1000)]
...: df = dask.dataframe.from_delayed(
...: [dask.delayed(read_file)(x, indx) for (indx, x) in enumerate(files_list)]
...: )
dataframe 预览最终看起来很相似，但如果我们深入查看高级任务图，我们会发现 dask 需要读入所有数据才能知道索引是什么样子！
In [16]: df.dask
Out[16]:
HighLevelGraph with 1001 layers.
<dask.highlevelgraph.HighLevelGraph object at 0x168bf6230>
0. read_file-b7aed020-1dc7-4872-a37d-d514b407a7d8
1. read_file-a0462606-999b-4af1-9977-acb562edab67
2. read_file-286df439-df34-4a5a-baf9-75dd0a5ae09b
3. read_file-4db8c178-a67e-4775-b117-228ac607f02f
4. read_file-a19d6144-5560-4da7-a1f5-8dc92b3ccf1c
# yeah... really there are 1000 of these...
998. read_file-d0cbd4a4-c255-4a77-a905-199bc289a0b5
999. read_file-45a80080-426a-48fd-8dcb-9ba7565307f1
1000. from-delayed-833eff6e232da1e10ca7221b961c21c1
更糟糕的是，每个 pd.read_parquet 使用默认的 pandas 读取行为，即假设数据可以装入内存并立即读取整个文件。 Pandas 不是 return 文件对象 - 它加载所有数据并 return 在 dask 甚至看到它之前是一个 DataFrame。
正因为如此，dask 基本上无法进入调度位，直到所有读取都已完成，并且在工作负载平衡、内存管理等方面几乎没有用处。它可以尝试通过执行第一个任务来获得 sneak-peek 的工作负载，但这仍然是对整个第一个文件的读取。
当我们开始尝试重新排列索引时，情况只会变得更糟。我不会在这里深入，但你明白了......