Editing CommunityData:Hyak Datasets (section)

=== Streaming parquet datasets ===
If the data you want to pull exceed available memory, you have a few options. 

One option is to just use [[CommunityData:Hyak_Spark|Spark]] which is likely a good option if you want to do large and complex joins or group-bys.  Downsides of Spark include issues of stability and complexity.  Spark is capable, can be fast, and can scale to many nodes, but it can also crash and be complex to program.

An alternative is to stream data from parquet using pyarrow. Pyarrow can load a large dataset one chunk at a time and you can turn these chunks into stream of rows.  The stream of rows will have the same order as the data on disk. In the example below the datasets are partitoned by author and the partitions are sorted so edits can be read one author at a time.  This is convenient as a starting point for building author-level variables. 

<syntaxhighlight lang='python'>
import pyarrow.dataset as ds
from itertools import groupby

# A pyarrow dataset abstracts reading, writing, or filtering a parquet file. It does not read data into memory. 
#dataset = ds.dataset(pathlib.Path('/gscratch/comdata/output/reddit_submissions_by_author.parquet/'), format='parquet', partitioning='hive')
dataset = ds.dataset('/gscratch/comdata/output/reddit_submissions_by_author.parquet', format='parquet')

# let's get all the comments to two subreddits:
subreddits_to_pull = ['seattlewa','seattle']

# instead of loading the data into a pandas dataframe all at once we can stream it.
scan_tasks = dataset.scan(filter = ds.field('subreddit').isin(subreddits_to_pull), columns=['id','subreddit','CreatedAt','author','ups','downs','score','subreddit_id','stickied','title','url','is_self','selftext'])

# simple function to execute scantasks and generate rows
def iterate_rows(scan_tasks):
    for st in scan_tasks:
        for rb in st.execute():
            df = rb.to_pandas()
            for t in df.itertuples():
                yield t

row_iter = iterate_rows(scan_tasks)

# now we can use python's groupby function to read one author at a time
# note that the same author can appear more than once since the record batches may not be in the correct order.
author_submissions = groupby(row_iter, lambda row: row.author)

count_dict = {}

for auth, posts in author_submissions:
    if auth in count_dict:
        count_dict[auth] = count_dict[auth] + 1
    else:
        count_dict[auth] = 1

# since it's partitioned and sorted by author, we get one group for each author 
any([ v != 1 for k,v in count_dict.items()])
</syntaxhighlight>