python - Fastparquet 似乎没有下压过滤器
问题描述
我使用 dask 的数据框to_parquet
方法创建了一个镶木地板文件,并将其fastparquet
用作引擎。使用 i 读取文件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'
</p>
所以字段是正确的。由于它们是时间序列数据,因此使用月份和日期对数据进行分区。数据总数为22815984
。现在我尝试使用过滤器关键字阅读镶木地板,但我得到了一个奇怪的行为。
# 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
还将过滤器更改为不存在的字段,不会引发任何异常,所以这也很奇怪。关于镶木地板和过滤,我有什么遗漏吗?
Dask DataFrame Structure:
http_user dst dst_port http_req_method
npartitions=612
object object float64 object
... ... ... ...
... ... ... ...
... ... ... ... ...
... ... ... ...
Dask Name: read-parquet, 612 tasks
解决方案
包含该filters=
选项是为了在有意义的情况下进行优化,以避免考虑确定不包含任何有效数据的数据部分。
在文档中:
这仅实现行组(分区)级过滤,即防止加载某些数据块,并且仅当相关统计信息已包含在元数据中时。
例如,如果您有一组行组,其中感兴趣的列单调增加,那么该列上的过滤器可能能够排除许多行组(也称为分区)。另一方面,如果每个行组都包含该列范围内的值,那么这种过滤器将产生任何效果。
data[raw_data.http_req_method == 'GET']
这做了一些不同的事情:现在每个行组都作为一个分区加载,然后在工作人员的内存中进行过滤。只有在您对索引进行过滤的特殊情况下,Dask 可能只能加载某些分区。
如果您想要优化,但您的数据不是结构化的分区边界与您的过滤条件完全一致,您将需要使用这两种方法。
如果您认为文档字符串更清晰,请提出问题。
推荐阅读
- apache-flink - 如何在 Apache flink KeyedBroadCastProcessFunction 中处理异常
- android - 改造错误:- 预期 BEGIN_ARRAY 但在路径 $ 处是 BEGIN_OBJECT
- python-3.x - 在 pipenv 中安装 twisted 时出错
- angular - Bitbucket Angular npm 运行构建失败
- c++ - 如何将 ostringstream 十六进制字符串字符对转换为单个 unit8_t 等效二进制值
- nginx - 目录的内容配置
- django - Django 管理员显示创建的实例,但在 DRF 视图中看不到它
- javascript - JavaScript 等待承诺
- javascript - 在 React 应用程序中使用 useState 更新路由器值时出现太多重新渲染错误
- java - 用于读取的带有 Java 的 Spark Cassandra 连接器