Add parallel upload for BulkImportWriter

[chezou]

This patch introduces parallel upload capability to BulkImportWriter.

It makes a chunk of 10,000 records and uploads in parallel. Also, it reduces memory consumption for msgpack format uploading.

Upload speed becomes x4.4 faster, and memory consumption can be controlled by using Temporary file and chunk_record_size number.

Dummy data creation for benchmark

>>> import numpy as np; import pandas as pd
>>> def fake_data(n):
...    users = np.random.choice([0., 1., 2.], (n, 1))
...    items = np.random.choice([0., 1., 2.], (n, 1))
...    weight = np.random.rand(n,1)
...    return np.concatenate((users, items, weight), axis=1)
...
>>> d1 = fake_data(10_000_000)
>>> df = pd.DataFrame(d1, columns=["users", "items", "scores"])
>>> import pytd; import os
>>> client=pytd.Client(database="aki", apikey=os.environ["TD_API_KEY"])

Upload with a single thread

>>> import time
>>> s=time.time()
>>> client.load_table_from_dataframe(df, "aki.pytd_bi_test", writer="bulk_import", if_exists="overwrite", fmt="msgpack", max_workers=1, chunk_record_size=10_000_000)
>>> print(f"elapsed time:{time.time() - s} sec")
uploading data converted into a msgpack file
uploaded data in 64.06 sec
performing a bulk import job
[job id 2172780406] imported 10000000 records.
elapsed time:281.57144117355347 sec

Upload with 6 threads

>>> import pytd; import os
>>> import time
>>> s=time.time()
>>> client.load_table_from_dataframe(df, "aki.pytd_bi_test", writer="bulk_import", if_exists="overwrite", fmt="msgpack", max_workers=6, chunk_record_size=1_000_000)
>>> print(f"elapsed time:{time.time() - s} sec")
uploading data converted into a msgpack file
uploaded data in 14.56 sec
performing a bulk import job
[job id 2172780795] imported 10000000 records.
elapsed time:209.5141899585724 sec

[shroman]

Suggested change

	file_like : List of file like objects
	file_likes : List of file like objects

[shroman]

Suggested change

	def _bulk_import(self, table, file_like, if_exists, fmt="csv", max_workers=5):
	def _bulk_import(self, table, file_likes, if_exists, fmt="csv", max_workers=5):

[shroman]

Suggested change

	for i, fp in enumerate(file_like):
	for i, fp in enumerate(file_likes):

[shroman]

Suggested change

	fp = file_like[0]
	fp = file_likes[0]

[shroman]

did this #seek become unnecessary?

[chezou]

Yes, it's moved to here https://github.com/treasure-data/pytd/pull/134/files#diff-7e3490636bfbd0197f47ef9694662f2621beb294fb178b5185c8f257082c53e7R533
This is because to get file size bystream.tell() properly, #seek should happen later.

[shroman]

maybe I missed something, but where this 62 comes from?

[chezou]

I tried to get the file size from fp, but I gave up because it doesn't match with the actual file size in the _write_msgpack_stream() method.