0
605
VistasLa forma más rápida de escribir una matriz numpy en formato de flecha
Estoy buscando formas rápidas de almacenar y recuperar una matriz numpy usando pyarrow . Estoy bastante satisfecho con la recuperación. Toma menos de 1 segundo extraer columnas de mi archivo .arrow que contiene 1.000.000.000 enteros de dtype = np.uint16 .
import pyarrow as pa import numpy as np def write(arr, name): arrays = [pa.array(col) for col in arr] names = [str(i) for i in range(len(arrays))] batch = pa.RecordBatch.from_arrays(arrays, names=names) with pa.OSFile(name, 'wb') as sink: with pa.RecordBatchStreamWriter(sink, batch.schema) as writer: writer.write_batch(batch) def read(name): source = pa.memory_map(name, 'r') table = pa.ipc.RecordBatchStreamReader(source).read_all() for i in range(table.num_columns): yield table.column(str(i)).to_numpy() arr = np.random.randint(65535, size=(250, 4000000), dtype=np.uint16) %%timeit -r 1 -n 1 write(arr, 'test.arrow') >>> 25.6 s ± 0 ns per loop (mean ± std. dev. of 1 run, 1 loop each) %%timeit -r 1 -n 1 for n in read('test.arrow'): n >>> 901 ms ± 0 ns per loop (mean ± std. dev. of 1 run, 1 loop each) ¿Se puede mejorar la eficiencia de escribir en formato .arrow ? Además, probé np.save :
%%timeit -r 1 -n 1 np.save('test.npy', arr) >>> 18.5 s ± 0 ns per loop (mean ± std. dev. of 1 run, 1 loop each) Se ve un poco más rápido. ¿Podemos optimizar Apache Arrow para escribir mejor en formato .arrow ?
2 Respuestas
Responde la pregunta0
Puede darse el caso de que el problema de rendimiento se deba principalmente a la velocidad de E/S/disco. En este caso, no hay mucho que puedas mejorar.
Hice algunas pruebas en mi dispositivo. Los números que obtengo son diferentes a los suyos. Pero la conclusión es la misma, escribir es más lento que leer.
El archivo resultante es de 1,9 GB (2000023184 bytes):
$ ls test.arrow -l -rw-rw-r-- 1 0x26res 0x26res 2000023184 Nov 15 10:01 test.arrowEn el siguiente código, genero 1,9 GB de bytes aleatorios y los guardo, luego los comparo con el tiempo que tardó en guardar con la flecha:
import secrets data = b"\x00" + secrets.token_bytes(2000023184) + b"\x00" def write_bytes(data, name): with open(name, 'wb') as fp: fp.write(data) %%timeit -r 1 -n 1 write_bytes(data, 'test.bytes') >>> 2.29 s ± 0 ns per loop (mean ± std. dev. of 1 run, 1 loop each) %%timeit -r 1 -n 1 write(arr, 'test.arrow') >>> 2.52 s ± 0 ns per loop (mean ± std. dev. of 1 run, 1 loop each)En mi dispositivo, toma 2.52 segundos escribir los datos usando la flecha. Si intento escribir esa cantidad de bytes aleatorios, me lleva 2,29 segundos. Significa que la sobrecarga o flecha es aproximadamente el 10% del tiempo de escritura, por lo que no se puede hacer mucho para acelerarlo.
0
De hecho, parece ser algún tipo de limitación de mi RAM/IO/disco. Muy silenciosos... Disminuye mi escritura de 3 a 8 veces después de que arr supera los 200 millones de elementos y es por eso que estoy experimentando una caída de velocidad de 2,5 segundos a 20. Me encantaría saber si esto se puede resolver en pyarrow .
def pyarrow_write_arrow_Batch(arr, name): arrays = [pa.array(col) for col in arr] names = [str(i) for i in range(len(arrays))] batch = pa.RecordBatch.from_arrays(arrays, names=names) with pa.OSFile(name, 'wb') as sink: with pa.RecordBatchStreamWriter(sink, batch.schema) as writer: writer.write_batch(batch) %matplotlib notebook import benchit benchit.setparams(environ='notebook') benchit.setparams(rep=5) arr = np.random.randint(65535, size=(int(1e9),), dtype=np.uint16) size = [4, 8, 12, 20, 32, 48, 64, 100, 160, 256, 400, 600, 1000] def pwa_Batch_10000(arr, name): return pyarrow_write_arrow_Batch(arr.reshape(-1, 10000), name) def pwa_Batch_100000(arr, name): return pyarrow_write_arrow_Batch(arr.reshape(-1, 100000), name) def pwa_Batch_1000000(arr, name): return pyarrow_write_arrow_Batch(arr.reshape(-1, 1000000), name) def pwa_Batch_4000000(arr, name): return pyarrow_write_arrow_Batch(arr.reshape(-1, 4000000), name) fns = [pwa_Batch_10000, pwa_Batch_100000, pwa_Batch_1000000, pwa_Batch_4000000] in_ = {s: (arr[:s*int(1e6)], 'test.arrow') for s in size} t = benchit.timings(fns, in_, multivar=True, input_name='Millions of items') t.plot(logx=True, figsize=(8,4), fontsize=10) 