defaults = sc.broadcast(dict(map(lambda x: (''.join(concat(x)), 0.0),
    product(product('HNL', 'NL', 'HNL'), repeat=2))))
rdd = sc.parallelize(['500, HNL, LNH, NLH, HNL',
    '600, HNN, NNN, NNN, HNN, LNH'])

def process(line):
    """
    >>> process("000, HHH, LLL, NNN")
    ('000', {'LLLNNN': 0.5, 'HHHLLL': 0.5})
    """
bits = line.split(', ')
transactions = bits[1:]
n = len(transactions) - 1
frequencies = pipe(sliding_window(2, transactions), map(lambda p: ''.join(p
    )), Counter, lambda cnt: {k: (v / n) for k, v in list(cnt.items())})
return bits[0], frequencies

def store_partition(iter):
    for k, v in iter:
        db_insert(k, merge([defaults.value, v]))
rdd.map(process).foreachPartition(store_partition)
Algorithmic / coding help for a PySpark markov model