collective/cdsc_reddit
13
0
Fork 0
Code

clean up comments in streaming example.

Browse Source
This commit is contained in:
Nate E TeBlunthuis 2020-07-07 12:28:57 -07:00
parent 7d0e020f9d
commit 06fd99e7cd
1 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
examples/pyarrow_streaming.py
View File

@ -1,17 +1,17 @@
pimport pyarrow.dataset as ds import pyarrow.dataset as ds
from itertools import chain, groupby, islice from itertools import groupby
# A pyarrow dataset abstracts reading, writing, or filtering a parquet file. It does not read dataa into memory. # A pyarrow dataset abstracts reading, writing, or filtering a parquet file. It does not read dataa into memory.
#dataset = ds.dataset(pathlib.Path('/gscratch/comdata/output/reddit_submissions_by_subreddit.parquet/'), format='parquet', partitioning='hive')
dataset = ds.dataset('/gscratch/comdata/output/reddit_submissions_by_author.parquet', format='parquet') dataset = ds.dataset('/gscratch/comdata/output/reddit_submissions_by_author.parquet', format='parquet')
# let's get all the comments to two subreddits: # let's get all the comments to two subreddits:
subreddits_to_pull = ['seattlewa','seattle'] subreddits_to_pull = ['seattlewa','seattle']
# instead of loading the data into a pandas dataframe all at once we can stream it. This lets us start working with it while it is read. # 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']) 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 create a stream of rows # simple function to execute scantasks and generate rows
def iterate_rows(scan_tasks): def iterate_rows(scan_tasks):
for st in scan_tasks: for st in scan_tasks:
for rb in st.execute(): for rb in st.execute():