> ## Documentation Index > Fetch the complete documentation index at: https://anaconda.com/docs/llms.txt > Use this file to discover all available pages before exploring further. # Load tabular data from cloud storage To read partitioned `.parquet` files from cloud storage into memory on remote workers quickly, combine the following three things: * Load data from S3 directly to memory with Metaflow's optimized S3 client (`metaflow.S3`), which reaches tens of gigabits per second or more. * Decode the Parquet data efficiently with Apache Arrow. * Keep the result in Arrow's in-memory tables, which are interoperable with modern data tools without additional copies, speeding up processing and avoiding unnecessary memory overhead. ### Cloud to table Before writing a Metaflow flow, look at how to use the [Metaflow S3 client](https://docs.metaflow.org/scaling/data) with [Apache Arrow](https://arrow.apache.org/). The main steps are using the [`metaflow.S3.get_many` function](https://docs.metaflow.org/api/S3#S3.get_many) to parallelize the retrieval of the `.parquet` file partitions, loading the bytes into memory on the worker instance, and decoding the bytes into a `pyarrow.Table` object. ```python theme={null} from metaflow import S3 import pyarrow.parquet as pq import pyarrow from concurrent.futures import ThreadPoolExecutor import multiprocessing ``` ```python theme={null} # Instantiate the Metaflow S3 client context s3 = S3() # Set the URL of an S3 bucket containing .parquet files url = "s3:///investment_ids" ``` Comments Replace \ with the name of your S3 bucket. The bucket in this example contains the [Ubiquant investment dataset](https://github.com/ubiquant), partitioned as `.parquet` files. To check metadata about what exists in the S3 URL of interest without actually downloading the files, use [`metaflow.S3.list_recursive`](https://docs.metaflow.org/scaling/data#listing-objects-in-s3): ```python theme={null} files = list(s3.list_recursive([url])) total_size = sum(f.size for f in files) / 1024**3 print("Loading%2.1dGB of data partitioned across %d files." % (total_size, len(files))) ``` ```text theme={null} Loading 7GB of data partitioned across 3579 files. ``` ```python theme={null} # Download the files in parallel loaded = s3.get_many([f.url for f in files]) ``` Notice the loaded files are in temporary storage in `./metaflow.s3.foobar`: ```python theme={null} print(len(loaded)) print(loaded[0]) print(loaded[0].path) ``` ```text theme={null} 3579 /investment_ids/0.parquet (1260224 bytes, local)> ./metaflow.s3.v_cz59co/9946232270752e97d9247ed2907154d2ea0b8841-0_parquet-whole ``` ```python theme={null} local_tmp_file_paths = [f.path for f in loaded] ``` In another set of parallel processes, read the PyArrow tables from bytes and then concatenate them: ```python theme={null} with ThreadPoolExecutor(max_workers = multiprocessing.cpu_count()) as exe: tables = exe.map(lambda f: pq.read_table(f, use_threads=False), local_tmp_file_paths) table = pyarrow.concat_tables(tables) ``` ```python theme={null} print("Table has %d rows and%2.1dGB bytes in memory." % (table.shape[0], table.nbytes / 1024**3)) ``` ```text theme={null} Table has 3141410 rows and 7GB bytes in memory. ``` ```python theme={null} # Close the S3 connection s3.close() ``` ### Performance benefits scale with instance size Using the basic pattern described above, you can write Metaflow flows that scale this fast data speedup on cloud instances. This workflow organizes the same operations presented in the previous section in a Metaflow flow. Notice that the `data_processing` step is annotated with `@batch(..., use_tmpfs=True, ...)`. The `tmpfs` feature extends the resources you request, because it allows you to use memory on the Batch instance to instantiate a temporary file system. This makes the cloud-to-table workflow significantly faster and does not require using the local file system to temporarily store the `.parquet` bytes. The benefits of this workflow scale with the number of processors, available RAM, and I/O throughput of the machine you are loading a table on, so use an instance that can fit your entire Arrow table in memory to get maximal benefits. To get a sense of how fast this workflow can get, read [Fast Data: Loading Tables From S3 At Lightning Speed](https://www.anaconda.com/blog/metaflow-fast-data). ```py title="fast_data_processing.py" expandable theme={null} from metaflow import Parameter, FlowSpec, step, S3, batch, conda from time import time class FastDataProcessing(FlowSpec): url = Parameter( "data", default="s3:///investment_ids", help="S3 prefix to Parquet files") @step def start(self): self.next(self.data_processing) @conda( libraries={ "pandas": "2.0.1", "pyarrow": "11.0.0" }, python="3.10.10" ) @batch(memory=32000, cpu=8, use_tmpfs=True, tmpfs_size=16000) @step def data_processing(self): import pyarrow.parquet as pq import pyarrow from concurrent.futures import ThreadPoolExecutor import multiprocessing with S3() as s3: # Check metadata about what is in the S3 URL of interest. files = list(s3.list_recursive([self.url])) total_size = sum(f.size for f in files) / 1024**3 msg = "Loading%2.1dGB of data across %d files." print(msg % (total_size, len(files))) # Download the parquet files in parallel. loaded = s3.get_many([f.url for f in files]) local_tmp_file_paths = [f.path for f in loaded] # Read the PyArrow tables from bytes and concatenate them. n_threads = multiprocessing.cpu_count() with ThreadPoolExecutor(max_workers = n_threads) as exe: tables = exe.map( lambda f: pq.read_table(f, use_threads=False), local_tmp_file_paths ) table = pyarrow.concat_tables(tables) msg = "Table has %d rows and%2.1dGB bytes in memory." print(msg % (table.shape[0], table.nbytes / 1024**3)) self.next(self.end) @step def end(self): pass if __name__ == "__main__": FastDataProcessing() ``` Comments Replace \ with the name of your S3 bucket containing partitioned `.parquet` files. ```bash theme={null} python fast_data_processing.py --environment=conda run ``` ```text expandable theme={null} Workflow starting (run-id 199435): [199435/start/1097414 (pid 71466)] Task is starting. [199435/start/1097414 (pid 71466)] Task finished successfully. [199435/data_processing/1097415 (pid 71475)] Task is starting. [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Task is starting (status SUBMITTED)... [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Task is starting (status RUNNABLE)... [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Task is starting (status RUNNABLE)... [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Task is starting (status RUNNABLE)... [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Task is starting (status RUNNABLE)... [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Task is starting (status STARTING)... [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Task is starting (status RUNNING)... [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Setting up task environment. [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Downloading code package... [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Code package downloaded. [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Task is starting. [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Bootstrapping environment... [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Environment bootstrapped. [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Loading 7GB of data across 3579 files. [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Table has 3141410 rows and 7GB bytes in memory. [199435/data_processing/1097415 (pid 71475)] [b731b181-128d-4e9b-9ed7-dd88e7f6cf26] Task finished with exit code 0. [199435/data_processing/1097415 (pid 71475)] Task finished successfully. [199435/end/1097416 (pid 71524)] Task is starting. [199435/end/1097416 (pid 71524)] Task finished successfully. Done! ```