Можно ли читать метаданные паркета из Dask? - PullRequest
Новая
       91

Можно ли читать метаданные паркета из Dask?

0 голосов
/ 03 октября 2019

У меня есть тысячи файлов паркета, которые мне нужно обработать. Перед обработкой файлов я пытаюсь получить различную информацию о файлах, используя метаданные паркета, такие как количество строк в каждом разделе, минимумы, максимумы и т. Д.

Я попытался прочитать метаданные с помощью dask. отложено в надежде распределить задачи сбора метаданных по моему кластеру, но это, похоже, приводит к нестабильности в Dask. См. Пример фрагмента кода и ошибку тайм-аута узла ниже.

Есть ли способ прочитать метаданные паркета из Dask? Я знаю, что функция Dask "read_parquet" имеет опцию "collect_statistics", которую вы можете установить в false, чтобы ускорить чтение файла. Но я не вижу способа получить доступ ко всем метаданным / статистике паркета, если для него установлено значение true.

Пример кода: 


@dask.delayed
def get_pf(item_to_read):
     pf = fastparquet.ParquetFile(item_to_read)
     row_groups = pf.row_groups.copy()
     all_stats = pf.statistics.copy()
     col = pf.info['columns'].copy()
     return [row_groups, all_stats, col]

stats_arr = get_pf(item_to_read)

Пример ошибки: 

2019-10-03 01:43:51,202 - INFO - 192.168.0.167 - distributed.worker - ERROR - Worker stream died during communication: tcp://192.168.0.223:34623

2019-10-03 01:43:51,203 - INFO - 192.168.0.167 - Traceback (most recent call last):

2019-10-03 01:43:51,204 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/distributed/comm/core.py", line 218, in connect

2019-10-03 01:43:51,206 - INFO - 192.168.0.167 -     quiet_exceptions=EnvironmentError,

2019-10-03 01:43:51,207 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/tornado/gen.py", line 729, in run

2019-10-03 01:43:51,210 - INFO - 192.168.0.167 -     value = future.result()

2019-10-03 01:43:51,211 - INFO - 192.168.0.167 - tornado.util.TimeoutError: Timeout

2019-10-03 01:43:51,212 - INFO - 192.168.0.167 -

2019-10-03 01:43:51,213 - INFO - 192.168.0.167 - During handling of the above exception, another exception occurred:

2019-10-03 01:43:51,214 - INFO - 192.168.0.167 -

2019-10-03 01:43:51,215 - INFO - 192.168.0.167 - Traceback (most recent call last):

2019-10-03 01:43:51,217 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/distributed/worker.py", line 1841, in gather_dep

2019-10-03 01:43:51,218 - INFO - 192.168.0.167 -     self.rpc, deps, worker, who=self.address

2019-10-03 01:43:51,219 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/tornado/gen.py", line 729, in run

2019-10-03 01:43:51,220 - INFO - 192.168.0.167 -     value = future.result()

2019-10-03 01:43:51,222 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/tornado/gen.py", line 736, in run

2019-10-03 01:43:51,223 - INFO - 192.168.0.167 -     yielded = self.gen.throw(*exc_info)  # type: ignore

2019-10-03 01:43:51,224 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/distributed/worker.py", line 3029, in get_data_from_worker

2019-10-03 01:43:51,225 - INFO - 192.168.0.167 -     comm = yield rpc.connect(worker)

2019-10-03 01:43:51,640 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/tornado/gen.py", line 729, in run

2019-10-03 01:43:51,641 - INFO - 192.168.0.167 -     value = future.result()

2019-10-03 01:43:51,643 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/tornado/gen.py", line 736, in run

2019-10-03 01:43:51,644 - INFO - 192.168.0.167 -     yielded = self.gen.throw(*exc_info)  # type: ignore

2019-10-03 01:43:51,645 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/distributed/core.py", line 866, in connect

2019-10-03 01:43:51,646 - INFO - 192.168.0.167 -     connection_args=self.connection_args,

2019-10-03 01:43:51,647 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/tornado/gen.py", line 729, in run

2019-10-03 01:43:51,649 - INFO - 192.168.0.167 -     value = future.result()

2019-10-03 01:43:51,650 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/tornado/gen.py", line 736, in run

2019-10-03 01:43:51,651 - INFO - 192.168.0.167 -     yielded = self.gen.throw(*exc_info)  # type: ignore

2019-10-03 01:43:51,652 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/distributed/comm/core.py", line 230, in connect

2019-10-03 01:43:51,653 - INFO - 192.168.0.167 -     _raise(error)

2019-10-03 01:43:51,654 - INFO - 192.168.0.167 -   File "/usr/local/lib/python3.7/dist-packages/distributed/comm/core.py", line 207, in _raise

2019-10-03 01:43:51,656 - INFO - 192.168.0.167 -     raise IOError(msg)

2019-10-03 01:43:51,657 - INFO - 192.168.0.167 - OSError: Timed out trying to connect to 'tcp://192.168.0.223:34623' after 10 s: connect() didn't finish in time

1 Ответ

0 голосов
/ 03 октября 2019

dd.read_parquet занимает много времени? Если нет, то вы можете использовать любую стратегию для чтения в клиенте.

Если в корневом каталоге данных содержится один файл _metadata, вы можете просто открыть его с помощью fastparquet,это именно то, что сделал бы Даск. Он содержит все детали всех частей данных.

Нет особой причины, по которой распространение чтения метаданных должно быть проблемой, но вы должны знать, что в некоторых случаях общее количество элементов метаданных может составлять в целомсущественный размер.

...