python - Fastparquet 似乎并没有推倒过滤器

Question

我使用 dask 的 dataframe to_parquet 方法创建了一个 parquet 文件，并使用 fastparquet 作为引擎。使用 fastparquet.ParquetFile 读取文件，我得到以下信息。

from fastparquet import ParquetFile
file = ParquetFile('data/raw_data_fastpar.par/')
file.dtypes
OrderedDict([(u'@timestamp', dtype('<M8[ns]')),
         (u'@version', dtype('O')),
         (u'_id', dtype('O')),
         (u'browser_build', dtype('O')),
         (u'browser_device', dtype('O')),
         (u'browser_major', dtype('float64')),
         (u'browser_minor', dtype('float64')),
         (u'browser_name', dtype('O')),
         (u'browser_os', dtype('O')),
         (u'browser_os_name', dtype('O')),
         (u'dst', dtype('O')),
         (u'dst_port', dtype('float64')),
         (u'http_req_header_contentlength', dtype('O')),
         (u'http_req_header_host', dtype('O')),
         (u'http_req_header_referer', dtype('O')),
         (u'http_req_header_useragent', dtype('O')),
         (u'http_req_headers', dtype('O')),
         (u'http_req_method', dtype('O')),
         (u'http_req_secondleveldomain', dtype('O')),
         (u'http_req_url', dtype('O')),
         (u'http_req_version', dtype('O')),
         (u'http_resp_code', dtype('O')),
         (u'http_resp_header_contentlength', dtype('O')),
         (u'http_resp_header_contenttype', dtype('O')),
         (u'http_resp_headers', dtype('O')),
         (u'http_user', dtype('O')),
         (u'received_from', dtype('O')),
         (u'redis_db', dtype('O')),
         (u'src', dtype('O')),
         (u'src_port', dtype('float64')),
         (u'type', dtype('O')),
         (u'month', u'category'),
         (u'day', u'category')])


file.schema.text
u'- schema: \n
| - @timestamp: INT64, TIMESTAMP_MICROS, OPTIONAL\n
| - @version: BYTE_ARRAY, UTF8, OPTIONAL\n
| - _id: BYTE_ARRAY, UTF8, OPTIONAL\n
| - browser_build: BYTE_ARRAY, UTF8, OPTIONAL\n
| - browser_device: BYTE_ARRAY, UTF8, OPTIONAL\n
| - browser_major: DOUBLE, OPTIONAL\n
| - browser_minor: DOUBLE, OPTIONAL\n
| - browser_name: BYTE_ARRAY, UTF8, OPTIONAL\n
| - browser_os: BYTE_ARRAY, UTF8, OPTIONAL\n
| - browser_os_name: BYTE_ARRAY, UTF8, OPTIONAL\n
| - dst: BYTE_ARRAY, UTF8, OPTIONAL\n
| - dst_port: DOUBLE, OPTIONAL\n
| - http_req_header_contentlength: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_req_header_host: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_req_header_referer: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_req_header_useragent: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_req_headers: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_req_method: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_req_secondleveldomain: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_req_url: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_req_version: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_resp_code: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_resp_header_contentlength: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_resp_header_contenttype: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_resp_headers: BYTE_ARRAY, UTF8, OPTIONAL\n
| - http_user: BYTE_ARRAY, UTF8, OPTIONAL\n
| - received_from: BYTE_ARRAY, UTF8, OPTIONAL\n
| - redis_db: BYTE_ARRAY, UTF8, OPTIONAL\n
| - src: BYTE_ARRAY, UTF8, OPTIONAL\n
| - src_port: DOUBLE, OPTIONAL\n  
| - type: BYTE_ARRAY, UTF8, OPTIONAL'

​

所以这些字段是正确的。由于它们是时间序列数据，因此使用月份和日期对数据进行分区。数据总数为22815984。现在，我尝试使用过滤器关键字读取 Parquet ，但出现了奇怪的行为。

# this works
import datetime
since = datetime.datetime(year=2018, month=10, day=1)
filters = [('@timestamp', '>', np.datetime64(since)),]

raw_data = dd.read_parquet('data/raw_data_fastpar.par/', engine='fastparquet', columns=['http_user', 'dst', 'dst_port', 'http_req_method'], filters=filters)

raw_data.count().compute()

http_user          3835971
dst                3835971
dst_port           3835971
http_req_method    3835971
dtype: int64

这是正确的，并且过滤已被按下。当我将过滤器更改为另一个字段时，

filters = [('http_req_method', '=', 'GET'),]

它取回所有数据

http_user          22815984
dst                22815984
dst_port           22815984
http_req_method    22815984
dtype: int64

手动执行，它有效:

raw_data = dd.read_parquet('data/raw_data_fastpar.par/', engine='fastparquet', columns=['http_user', 'dst', 'dst_port', 'http_req_method'])
raw_data.loc[raw_data.http_req_method == 'GET'].count().compute()
http_user          14407709
dst                14407709
dst_port           14407709
http_req_method    14407709
dtype: int64

还将过滤器更改为不存在的字段，不会引发任何异常，所以这也很奇怪。关于 Parquet 和过滤，我是否遗漏了什么？

Dask DataFrame Structure:
    http_user   dst     dst_port    http_req_method
npartitions=612                 
    object      object  float64         object
    ...         ...     ...             ...
    ...         ...     ...             ...     
... ...         ...     ...             ...
    ...         ...     ...             ...
Dask Name: read-parquet, 612 tasks

Answer 1

包含 filters= 选项是为了在有意义的情况下进行优化，以避免考虑肯定不包含任何有效数据的数据部分。

在 docs :

This implements row-group (partition) -level filtering only, i.e., to prevent the loading of some chunks of the data, and only if relevant statistics have been included in the metadata.

例如，如果您有一组行组，其中感兴趣的列单调递增，则该列上的过滤器可能能够排除许多行组(也称为分区)。另一方面，如果每个行组都包含该列范围内的值，则这种过滤器将产生任何效果。

data[raw_data.http_req_method == 'GET']

这做了一些不同的事情:现在每个行组都作为分区加载，然后在工作人员的内存中进行过滤。仅在您对索引进行过滤的特殊情况下，Dask 可能只能加载某些分区。

如果您想要优化，但您的数据结构不使得分区边界与您的过滤条件完全一致，则您将需要使用这两种方法。

如果您认为文档字符串可以更清晰，请提出问题。