<jupyter_start><jupyter_text>Code to read raw reddit json files and convert them to parquet- working with parquet files is much faster than json ones First imports- we neet to get an `sqlContext` from spark to do our work in- The `sc` (spark context) variable is already set for us, we can just use it<jupyter_code>from pyspark.sql import SQLContext
from pyspark.sql.functions import *

sqlC = SQLContext(sc)<jupyter_output><empty_output><jupyter_text>Read in json data- all the comments data is in /var/reddit    - the comments data is in files by month and begins with `RC`    - the submissions data (things comments reply to, "first posts" in each thread) is in files by month and begins with `RS`- For all comments, this takes about half an hour on UM's flux hadoop cluster.    - you can make it faster if you use just the files for a specific year or month.<jupyter_code># all of reddit:
#reddit = sqlC.read.json("/var/reddit/RC_*")

#one month
reddit_raw = sqlC.read.json("/var/reddit/RC_2010-02")
print('Done!')
reddit_raw.columns
reddit_raw.printSchema()
reddit_raw.count()<jupyter_output><empty_output><jupyter_text>Save data to a faster format- spark works much faster on .parquet formatted files. What takes minutes or hours in json is seconds in parquet.<jupyter_code>reddit_raw.write.parquet('2010_02_comments.parquet', mode='overwrite')
print('done')
reddit = sqlC.read.parquet('2010_02_comments.parquet')
reddit.count()<jupyter_output><empty_output><jupyter_text>Speed difference- Difference grows exponentially, really appears in larger data or more complicated operations<jupyter_code>%timeit reddit_raw.count()
%timeit reddit.count()<jupyter_output><empty_output><jupyter_text>Basic operatons<jupyter_code>#dataframe style
reddit[['id', 'author', 'subreddit', 'body']].show()
#sql style
reddit.select('id', 'author', 'subreddit', 'body').show()
#dataframe style
reddit[reddit.subreddit == 'politics'].count()
#sql style
reddit.filter(reddit.subreddit == 'politics').count()
reddit[reddit.body.contains('kitten') | reddit.body.contains('Kitten')].count()<jupyter_output><empty_output><jupyter_text>built-in pyspark functionse.g. lower()<jupyter_code>reddit[lower(reddit.body).contains('kitten')].count()<jupyter_output><empty_output><jupyter_text>Changing data- We can't modify or assign things to pyspark dataframes like normal- There are special functions for that<jupyter_code>reddit = reddit.withColumnRenamed('id', 'comment_id')
reddit.columns
reddit = reddit.withColumn('date', from_unixtime('created_utc'))
reddit[['comment_id', 'subreddit', 'created_utc', 'date']].show(5)
reddit = reddit.withColumn('body', regexp_replace('body', '\t', ' '))<jupyter_output><empty_output><jupyter_text>More fun examples<jupyter_code>reddit.groupBy('author').count().sort(desc('count')).show()
reddit = reddit[reddit.author != '[deleted]']
reddit.groupBy('subreddit').count().sort(desc('count')).show()
reddit.groupBy('subreddit', 'author').count().sort(desc('count')).show()
tmp = reddit.groupBy('subreddit', 'author').count()
tmp = tmp.groupby('subreddit').mean()
tmp.sort(desc('avg(count)')).show()
tmp.withColumnRenamed('avg(count)', 'comments_per_user').show()
tmp.select('subreddit', col('avg(count)').alias('comments_per_user2')).show()<jupyter_output><empty_output><jupyter_text>Joins for networks- making a network of authors who participate in the same subreddits by using a self-join<jupyter_code>tmp = reddit[['subreddit', 'author']]
author_network = tmp.join(tmp[['subreddit', col('author').alias('author2')]], on='subreddit')
author_network = author_network.drop_duplicates()
author_network.show()<jupyter_output><empty_output><jupyter_text>for sets- Sometimes you want everyone who *isn't* in one group, and we have ant-join for that<jupyter_code>funny_users = reddit[reddit.subreddit == 'funny'][['author']].distinct()
funny_users.count()
science_users = reddit[reddit.subreddit == 'science'][['author']].distinct()
science_users.count()
#users in both subreddits
science_users.join(funny_users, on='author', how='inner').count()
#users in only science
science_users.join(funny_users, on='author', how='left_anti').count()
funny_scientists = reddit[reddit['subreddit'] == 'science']
funny_scientists = funny_scientists.join(funny_users, on='author', how='inner')
funny_scientists[['body']].show(10, truncate=False)
boring_scientists = reddit[reddit['subreddit'] == 'science']
boring_scientists = boring_scientists.join(funny_users, on='author', how='left_anti')
boring_scientists[['body']].show(10, truncate=False)
funny_scientists.count()
fs_df = funny_scientists.toPandas()
fs_df.head()
fs_df.to_csv('funny_scientists.tsv', sep='\t', index=False, encoding='UTF-8')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14851694915254238, 'fraction_numerical': 0.005508474576271186, 'mean_word_length': 8.009541984732824, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '353990', 'license': 'CC-BY-3.0', 'n_tokens_mistral': 1557, 'n_tokens_neox': 1595, 'n_words': 357, 'path': 'data/data_camp_pyspark_demo.ipynb', 'repo_name': 'ICOSBigDataCamp/2019-summer-camp', 'size': '4720'}