ການຖ່າຍທອດຂໍ້ມູນຖັນກັບ Apache Arrow

ການແປພາສາຂອງບົດຄວາມໄດ້ຖືກກະກຽມໂດຍສະເພາະສໍາລັບນັກສຶກສາຂອງຫຼັກສູດ ວິສະວະກອນຂໍ້ມູນ.

ໃນໄລຍະສອງສາມອາທິດຜ່ານມາ, ພວກເຮົາ ໜອງລີ ເພີ່ມໃສ່ Apache Arrow ຮູບ​ແບບ​ສະ​ຕ​ຣີມ​ຄູ່​, ເສີມ​ການ​ເຂົ້າ​ເຖິງ​ແບບ​ສຸ່ມ​ທີ່​ມີ​ຢູ່​ແລ້ວ / ຮູບ​ແບບ​ໄຟລ​໌ IPC​. ພວກເຮົາມີການປະຕິບັດ Java ແລະ C++ ແລະການຜູກມັດ Python. ໃນບົດຄວາມນີ້, ຂ້າພະເຈົ້າຈະອະທິບາຍວິທີການເຮັດວຽກຂອງຮູບແບບແລະສະແດງໃຫ້ເຫັນວິທີທີ່ທ່ານສາມາດບັນລຸການສົ່ງຂໍ້ມູນສູງຫຼາຍສໍາລັບ pandas DataFrame.

ການຖ່າຍທອດຂໍ້ມູນຖັນ

ຄໍາຖາມທົ່ວໄປທີ່ຂ້ອຍໄດ້ຮັບຈາກຜູ້ໃຊ້ Arrow ແມ່ນຄ່າໃຊ້ຈ່າຍສູງໃນການເຄື່ອນຍ້າຍຊຸດຂໍ້ມູນຕາຕະລາງຂະຫນາດໃຫຍ່ຈາກແຖວຫຼືຮູບແບບບັນທຶກເປັນຮູບແບບຄໍລໍາ. ສຳລັບຊຸດຂໍ້ມູນຫຼາຍກິກາໄບ, ການຖ່າຍທອດໃນໜ່ວຍຄວາມຈຳ ຫຼືໃນດິສກ໌ອາດເປັນເລື່ອງທີ່ໜັກໜ່ວງ.

ສໍາລັບການຖ່າຍທອດຂໍ້ມູນ, ບໍ່ວ່າຈະເປັນຂໍ້ມູນແຫຼ່ງແມ່ນແຖວຫຼືຖັນ, ທາງເລືອກຫນຶ່ງແມ່ນເພື່ອສົ່ງ batch ຂະຫນາດນ້ອຍຂອງແຖວ, ແຕ່ລະປະກອບດ້ວຍໂຄງຮ່າງຖັນພາຍໃນ.

ໃນ Apache Arrow, ຄໍເລັກຊັນຂອງຖັນຖັນໃນໜ່ວຍຄວາມຈຳທີ່ເປັນຕົວແທນຂອງຕາຕະລາງແມ່ນເອີ້ນວ່າຊຸດບັນທຶກ. ເພື່ອເປັນຕົວແທນຂອງໂຄງສ້າງຂໍ້ມູນດຽວຂອງຕາຕະລາງຢ່າງມີເຫດຜົນ, ທ່ານສາມາດເກັບກໍາຫຼາຍຊຸດຂອງບັນທຶກ.

ໃນຮູບແບບໄຟລ໌ "ການເຂົ້າເຖິງແບບສຸ່ມ" ທີ່ມີຢູ່ແລ້ວ, ພວກເຮົາຂຽນ metadata ທີ່ມີຕາຕະລາງຕາຕະລາງແລະຮູບແບບຕັນຢູ່ໃນຕອນທ້າຍຂອງໄຟລ໌, ເຊິ່ງຊ່ວຍໃຫ້ທ່ານສາມາດເລືອກເອົາຊຸດບັນທຶກຫຼືຄໍລໍາໃດໆຈາກຊຸດຂໍ້ມູນທີ່ມີລາຄາຖືກຫຼາຍ. ໃນ​ຮູບ​ແບບ​ການ​ສະ​ຕ​ຣີມ​, ພວກ​ເຮົາ​ສົ່ງ​ຊຸດ​ຂອງ​ຂໍ້​ຄວາມ​: schema​, ແລະ​ຫຼັງ​ຈາກ​ນັ້ນ​ຫນຶ່ງ​ຫຼື​ຫຼາຍ​ຊຸດ​ຂອງ​ການ​ບັນ​ທຶກ​.

ຮູບແບບທີ່ແຕກຕ່າງກັນເບິ່ງຄືຮູບນີ້:

ການຖ່າຍທອດຂໍ້ມູນໃນ PyArrow: ແອັບພລິເຄຊັນ

ເພື່ອສະແດງໃຫ້ທ່ານຮູ້ວ່າມັນເຮັດວຽກແນວໃດ, ຂ້ອຍຈະສ້າງຊຸດຂໍ້ມູນຕົວຢ່າງທີ່ເປັນຕົວແທນຂອງສາຍນ້ໍາດຽວ:

import time
import numpy as np
import pandas as pd
import pyarrow as pa

def generate_data(total_size, ncols):
    nrows = int(total_size / ncols / np.dtype('float64').itemsize)
    return pd.DataFrame({
        'c' + str(i): np.random.randn(nrows)
        for i in range(ncols)
    })	

ໃນປັດຈຸບັນ, ສົມມຸດວ່າພວກເຮົາຕ້ອງການຂຽນຂໍ້ມູນ 1 GB, ປະກອບດ້ວຍ 1 MB ແຕ່ລະ chunks, ສໍາລັບທັງຫມົດ 1024 chunks. ກ່ອນອື່ນ, ໃຫ້ສ້າງກອບຂໍ້ມູນ 1MB ທໍາອິດທີ່ມີ 16 ຖັນ:

KILOBYTE = 1 << 10
MEGABYTE = KILOBYTE * KILOBYTE
DATA_SIZE = 1024 * MEGABYTE
NCOLS = 16

df = generate_data(MEGABYTE, NCOLS)

ຫຼັງຈາກນັ້ນ, ຂ້າພະເຈົ້າປ່ຽນໃຫ້ເຂົາເຈົ້າ pyarrow.RecordBatch:

batch = pa.RecordBatch.from_pandas(df)

ຕອນນີ້ຂ້ອຍຈະສ້າງກະແສຜົນຜະລິດທີ່ຈະຂຽນໃສ່ RAM ແລະສ້າງ StreamWriter:

sink = pa.InMemoryOutputStream()
stream_writer = pa.StreamWriter(sink, batch.schema)

ຫຼັງຈາກນັ້ນ, ພວກເຮົາຈະຂຽນ 1024 chunks, ເຊິ່ງໃນທີ່ສຸດຈະປະກອບເປັນຊຸດຂໍ້ມູນ 1GB:

for i in range(DATA_SIZE // MEGABYTE):
    stream_writer.write_batch(batch)

ນັບຕັ້ງແຕ່ພວກເຮົາຂຽນໃນ RAM, ພວກເຮົາສາມາດໄດ້ຮັບນ້ໍາທັງຫມົດໃນ buffer ດຽວ:

In [13]: source = sink.get_result()

In [14]: source
Out[14]: <pyarrow.io.Buffer at 0x7f2df7118f80>

In [15]: source.size
Out[15]: 1074750744

ເນື່ອງຈາກວ່າຂໍ້ມູນນີ້ຢູ່ໃນຫນ່ວຍຄວາມຈໍາ, ການອ່ານ batches ຂອງ Arrow records ແມ່ນໄດ້ຮັບໂດຍການດໍາເນີນການສູນສໍາເນົາ. ຂ້ອຍເປີດ StreamReader, ອ່ານຂໍ້ມູນເຂົ້າໄປໃນ pyarrow.Tableແລະຫຼັງຈາກນັ້ນປ່ຽນໃຫ້ເຂົາເຈົ້າເປັນ DataFrame pandas:

In [16]: reader = pa.StreamReader(source)

In [17]: table = reader.read_all()

In [18]: table
Out[18]: <pyarrow.table.Table at 0x7fae8281f6f0>

In [19]: df = table.to_pandas()

In [20]: df.memory_usage().sum()
Out[20]: 1073741904

ທັງຫມົດນີ້, ແນ່ນອນ, ແມ່ນດີ, ແຕ່ທ່ານອາດຈະມີຄໍາຖາມ. ມັນເກີດຂຶ້ນໄວເທົ່າໃດ? ຂະຫນາດຂອງ chunk ມີຜົນກະທົບຕໍ່ການປະຕິບັດການດຶງຂໍ້ມູນ pandas DataFrame ແນວໃດ?

ປະສິດທິພາບການຖ່າຍທອດ

ເມື່ອຂະໜາດຂອງຊ່ອງສະຕຣີມມິງຫຼຸດລົງ, ຄ່າໃຊ້ຈ່າຍຂອງການສ້າງຄໍລຳ DataFrame ຕິດຕໍ່ກັນໃນ pandas ເພີ່ມຂຶ້ນເນື່ອງຈາກລະບົບການເຂົ້າເຖິງແຄດບໍ່ມີປະສິດທິພາບ. ນອກຈາກນີ້ຍັງມີບາງສ່ວນເກີນຈາກການເຮັດວຽກກັບໂຄງສ້າງຂໍ້ມູນ C ++ ແລະ arrays ແລະ buffers ຄວາມຊົງຈໍາຂອງພວກເຂົາ.

ສໍາລັບ 1 MB ດັ່ງຂ້າງເທິງ, ໃນຄອມພິວເຕີ້ຂອງຂ້ອຍ (Quad-core Xeon E3-1505M) ມັນປາກົດວ່າ:

In [20]: %timeit pa.StreamReader(source).read_all().to_pandas()
10 loops, best of 3: 129 ms per loop

ມັນ turns ໃຫ້ເຫັນວ່າ throughput ປະສິດທິພາບແມ່ນ 7.75 Gb / s ສໍາລັບການຟື້ນຟູ 1 GB DataFrame ຈາກ 1024 1 MB chunks. ຈະເກີດຫຍັງຂຶ້ນຖ້າພວກເຮົາໃຊ້ຕ່ອນໃຫຍ່ ຫຼືນ້ອຍກວ່າ? ນີ້ແມ່ນຜົນໄດ້ຮັບທີ່ທ່ານໄດ້ຮັບ:

ປະສິດທິພາບຫຼຸດລົງຢ່າງຫຼວງຫຼາຍຈາກ 256K ຫາ 64K chunks. ຂ້າ​ພະ​ເຈົ້າ​ຕົກ​ຕະ​ລຶງ​ທີ່ chunks 1MB ຖືກ​ປຸງ​ແຕ່ງ​ໄວ​ກ​່​ວາ 16MB chunks​. ມັນຄຸ້ມຄ່າທີ່ຈະເຮັດການສຶກສາຢ່າງລະອຽດກວ່າແລະເຂົ້າໃຈວ່ານີ້ແມ່ນການແຈກຢາຍປົກກະຕິຫຼືບາງສິ່ງບາງຢ່າງທີ່ກ່ຽວຂ້ອງ.

ໃນການປະຕິບັດໃນປະຈຸບັນຂອງຮູບແບບ, ຂໍ້ມູນບໍ່ໄດ້ຖືກບີບອັດໃນຫຼັກການ, ດັ່ງນັ້ນຂະຫນາດໃນຫນ່ວຍຄວາມຈໍາແລະ "ສາຍ" ແມ່ນປະມານດຽວກັນ. ການບີບອັດອາດຈະກາຍເປັນທາງເລືອກໃນອະນາຄົດ.

ຜົນໄດ້ຮັບ

ການຖ່າຍທອດຂໍ້ມູນຖັນສາມາດເປັນວິທີທີ່ມີປະສິດທິພາບໃນການໂອນຊຸດຂໍ້ມູນຂະໜາດໃຫຍ່ໄປຫາເຄື່ອງມືການວິເຄາະຖັນເຊັ່ນ: pandas ໃນຕ່ອນນ້ອຍໆ. ການບໍລິການຂໍ້ມູນທີ່ໃຊ້ບ່ອນເກັບຂໍ້ມູນແບບແຖວສາມາດໂອນ ແລະຖ່າຍທອດຂໍ້ມູນນ້ອຍໆທີ່ສະດວກກວ່າສຳລັບແຄດ L2 ແລະ L3 ຂອງໂປເຊດເຊີຂອງທ່ານ.

ລະຫັດເຕັມ

import time
import numpy as np
import pandas as pd
import pyarrow as pa

def generate_data(total_size, ncols):
    nrows = total_size / ncols / np.dtype('float64').itemsize
    return pd.DataFrame({
        'c' + str(i): np.random.randn(nrows)
        for i in range(ncols)
    })

KILOBYTE = 1 << 10
MEGABYTE = KILOBYTE * KILOBYTE
DATA_SIZE = 1024 * MEGABYTE
NCOLS = 16

def get_timing(f, niter):
    start = time.clock_gettime(time.CLOCK_REALTIME)
    for i in range(niter):
        f()
    return (time.clock_gettime(time.CLOCK_REALTIME) - start) / NITER

def read_as_dataframe(klass, source):
    reader = klass(source)
    table = reader.read_all()
    return table.to_pandas()
NITER = 5
results = []

CHUNKSIZES = [16 * KILOBYTE, 64 * KILOBYTE, 256 * KILOBYTE, MEGABYTE, 16 * MEGABYTE]

for chunksize in CHUNKSIZES:
    nchunks = DATA_SIZE // chunksize
    batch = pa.RecordBatch.from_pandas(generate_data(chunksize, NCOLS))

    sink = pa.InMemoryOutputStream()
    stream_writer = pa.StreamWriter(sink, batch.schema)

    for i in range(nchunks):
        stream_writer.write_batch(batch)

    source = sink.get_result()

    elapsed = get_timing(lambda: read_as_dataframe(pa.StreamReader, source), NITER)

    result = (chunksize, elapsed)
    print(result)
    results.append(result)

ແຫຼ່ງຂໍ້ມູນ: www.habr.com