Hi, I recently asked about saving datasets as artifacts. https://outerbounds-community.slack.com/archives/C02116BBNTU/p1680550636387349. I was previously working with pandas dataframes but after reading https://outerbounds-community.slack.com/archives/C02116BBNTU/p1638394866487100 and going through chapter 7 of the effective data science infrastructure book, decided to work with pyarrow tables. My question is when should you consider saving pyarrow tables as metaflow artifacts versus when to store them in S3 in a metaflow versioned bucket. I do see that in https://outerbounds.com/docs/chunk-df/, we can save pyarrow tables as artifacts but when should you refrain from doing that? Thanks again.

If the serialization cost of

self.

is too big, you can chunk the data to parquet files in the upstream task, and read them downstream using the cloud-to-table pattern in this blog. More context Calling

self.artifact=my_table

serializes the contents of

my_table

and pushes the bytes to cloud storage which can become relatively slow with big tables/dfs. The reason to use

self.

is to benefit from Metaflow's versioning across tasks run on different machines. For smallish datasets, there is a negligible cost to versioning artifacts like this and the benefits make debugging and monitoring much easier. But as the data size grows, and benefits of the fast data pattern kick in more, and the serialization cost of

self.

grows. By the way, "the cost" of serialization here is described above like an opportunity cost relative to the fast data pattern, since it is actually pretty fast to serialize and move artifacts within the same cloud - the thing that matters more is that the table is one big blob, as opposed to N parquet files that can be read/written/operated on in parallel.

Thanks for the quick response! I guess my next question would be when is there a way to determine before hand if the serialization cost is too high? For example, if I see my dataframe is going to be bigger than 5 GB, I use the metaflow s3 client where run=self to save the dataset. Is there something similar we can do with pyarrow tables?

I don't know a generally good heuristic on data size threshold. Usually I see parquet files get chunked to a size around .1GB - 1GB, so a few GBs seems like a reasonable place to start. The S3 pattern for going from pyarrow table in memory to chunked parquet files in S3 would be similar. You could open metaflow.S3 context and use pyarrow.dataset.write_dataset + s3.put_files. You could also use pyarrow S3 Filesystem as an argument to write_dataset.

Thanks @crooked-jordan-29960.

Hmm, not sure that using S3 context tmproot as base directory sounds like the right approach. 🤔 Definitely makes sense to avoid writing to disk. Using bytesIO (or ofc pyarrow) buffer is a nice pattern!

from metaflow import FlowSpec, step

class F(FlowSpec):

        @step
        def start(self):
                self.next(self.end)

        @step
        def end(self):
                from metaflow import S3
                import io
                buf = io.BytesIO()
                buf.write(b"How many bytes do you think this message is? Check in the S3 bucket metadata!")
                buf.seek(0)
                with S3(run=self) as s3:
                        url = s3.put('data', buf)
                        print(url)

if __name__ == '__main__':
        F()

Also for small dataframes or chunks AWSWrangler is nice!

import awswrangler as wr

wr.pandas.to_parquet(
    dataframe=df,
    path="s3://...",
    dataset=True,
    mode="overwrite",        # Could be append, overwrite or overwrite_partitions
    database="my_database",  # Optional, only with you want it available on Athena/Glue Catalog
    table="my_table",
    partition_cols=["PARTITION_COL_NAME"])

Thanks Eddie, I think what I've settled on is to convert the table into record batches that are of a certain chunksize, and for each batch write to parquet on BytesIO buffer and using s3.put