大数据处理dask的一些问题

Question

请看源码：

import dask.dataframe as dd

# 读取大数据集
df = dd.read_csv('../6/数据源/farming2.csv',assume_missing=True)

#数据类型转换
df['平均交易价格'] = dd.to_numeric(df['平均交易价格'], errors='coerce')
print(df['平均交易价格'].dtypes)
print(df.dtypes)
# 进行数据处理
mean_df = df['平均交易价格'].mean(numeric_only=True).compute()

请看报错

神奇的是，这列数据它就是数值型

所以这个报错很奇怪，其实目标就是为了找出“平均交易价格”这列数据的异常值由于数据量太大之前pandas一直跑不出结果就换了dask去处理，结果现在报错了，问了AI也解决不了。

以下附上数据集，感谢指点thx
[](通过网盘分享的文件：data
链接: https://pan.baidu.com/s/18YkogYGuOEHCsBC8NfioBQ?pwd=xtyd 提取码: xtyd)

Answer 1

在使用 Dask 进行大数据处理时，您可能会遇到一些问题，以下是一些建议和解决方案：

1. 数据类型转换问题：当您尝试将一列转换为数值类型时，如果数据中包含无法转换的值，可以使用 errors='coerce' 参数，这将会把无法转换的值设置为 NaN。例如：

   df['平均交易价格'] = dd.to_numeric(df['平均交易价格'], errors='coerce')
   

   这样可以确保转换过程中不会因为错误而中断。

2. 性能问题：如果您在处理数据时遇到性能瓶颈，可以尝试以下方法：

   • 确保您的数据块大小（blocksize）适中，以便能够有效利用内存。
   • 使用 map_partitions 方法来应用自定义函数，这可以提高处理速度。
   • 如果可能，避免在每个分区上进行重复的计算，而是尝试一次性计算所有需要的结果。

3. 读取大数据集：当您尝试读取一个非常大的数据集时，可以使用 assume_missing=True 参数来帮助 Dask 更好地推断数据类型，尤其是在数据集中有缺失值的情况下。

4. 内存优化：在处理大型数据集时，内存使用是一个关键问题。您可以使用 reduce_mem_usage 函数来优化内存使用，例如：

   def reduce_mem_usage(df):
       # ...
       return df
   

   这个函数会遍历数据框的每一列，并将数据类型转换为更小的类型，以减少内存占用。

5. 并行计算：Dask 的一大优势是能够并行处理任务。确保您的计算任务被正确地分散到多个核心上，可以通过 Client 类来设置并行计算的客户端：

   from dask.distributed import Client
   client = Client(n_workers=4)  # 根据您的核心数设置
   

   这样可以帮助您更有效地利用多核处理器。

6. 调试和可视化：如果您需要理解 Dask 任务的执行流程，可以使用 visualize 函数来生成任务图，这有助于调试和优化您的计算流程：

   df['平均交易价格'].mean().visualize(optimize_graph=True)
   

7. 数据预处理：在开始大规模计算之前，进行数据清洗和预处理可以减少后续计算的负担。

8. 选择合适的数据结构：Dask 提供了多种数据结构，如 Dask Array、DataFrame 和 Bag。根据您的具体需求选择合适的数据结构可以提高效率。