<jupyter_start><jupyter_text>Policy Laplace on SparkSpark implementation of Policy Laplace from Differentially Private Set Union [https://arxiv.org/abs/2002.09745]<jupyter_code>import pyspark
import numpy as np
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
spark = SparkSession.builder.getOrCreate()
from os import path
if not path.exists('clean_askreddit.csv'):
    if not path.exists('clean_askreddit.csv'):
        !pip install wget
        import wget
        zip_path = 'https://dp-test-datasets.s3.amazonaws.com/clean_askreddit.csv.zip'
        wget.download(zip_path)
    import zipfile
    with zipfile.ZipFile('clean_askreddit.csv.zip', 'r') as zip:
        zip.extractall('.')
    
filepath = "clean_askreddit.csv"
reddit = spark.read.load(filepath, format="csv", sep=",",inferSchema="true", header="true").dropna()<jupyter_output><empty_output><jupyter_text>Prepare Data for ProcessingLoad the data from file and tokenize.  This code can be any caller-specific tokenization routine, and is independent of differential privacy.  Output RDD should include one list of tokens per row, but can have multiple rows per user, and does not need to be odered in any way.  This stage can be combined with other n_grams (e.g. 2-grams, 3-grams) and persisted to feed to DPSU.<jupyter_code>import nltk

n_grams = 1
distinct = True

def tokenize(user_post):
    user, post = user_post
    tokens = post.split(" ")
    if n_grams > 1:
        tokens = list(nltk.ngrams(tokens, n_grams))
        tokens = ["_".join(g) for g in tokens]
    if distinct:
        tokens = list(set(tokens))
    return (user, tokens)
        
tokenized = reddit.select("author", "clean_text").rdd.map(tokenize).persist()<jupyter_output><empty_output><jupyter_text>Instantiate DPSU ProcessorCreate the object and pass in the privacy parameters.<jupyter_code>from policy_laplace import PolicyLaplace

epsilon = 3.0
delta = np.exp(-10)
alpha = 5.0
tokens_per_user = 500
prune_tail_below = None
num_partitions = 1

pl = PolicyLaplace(epsilon, delta, alpha, tokens_per_user, prune_tail_below, num_partitions)
# prune the tail
pruned = pl.prune_tail(tokenized)

# reservoir sample the input tokens
sampled = pl.reservoir_sample(pruned, distinct).persist()
counted = pl.process_partitions(sampled)
good = counted.filter(lambda row: pl.exceeds_threshold(row[1])).map(lambda row: row[0])

print("Retrieved {0} words from {1}".format(good.count(),counted.count()))
print(good.take(5))<jupyter_output>Retrieved 13253 words from 150591
["n't", 'baby', 'escape', 'kinds', 'alive']
{'fraction_non_alphanumeric': 0.11098130841121495, 'fraction_numerical': 0.01557632398753894, 'mean_word_length': 4.786036036036036, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 13, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '186879', 'license': 'MIT', 'n_tokens_mistral': 885, 'n_tokens_neox': 836, 'n_words': 230, 'path': 'data/Policy Laplace on Spark.ipynb', 'repo_name': 'AdrienRoyer/smartnoise-samples', 'size': '2568'}