PySpark (Python 2.7): How to flatten values after reduce
Question: I'm reading a multiline-record file using SparkContext.newAPIHadoopFile with a
customized delimiter. Anyway, I already prepared, reduced my data. But now I
want to add the key to every line (entry) again and then write it to a Apache
Parquet file, which is then stored into HDFS.

This figure should explain my problem. What I'm looking for is the red arrow
e.g. the last transformation before writing the file. Any idea? I tried
flatMap but then the timestamp and float-value resulted in different records.

[![PySpark
chain](http://i.stack.imgur.com/Vz9dp.png)](http://i.stack.imgur.com/Vz9dp.png)

The Python-Script can be [downloaded here](http://matthias-
heise.eu/stackoverflow/sample.py) and the sample [text file
here](http://matthias-heise.eu/stackoverflow/sample.txt). I'm using the
Python-Code within a Jupyter Notebook.


Answer: Simple list comprehension should be more than enough:

    
    
    from datetime import datetime
    
    
    def flatten(kvs):
        """
        >>> kvs = ("852-YF-008", [
        ... (datetime(2016, 5, 10, 0, 0), 0.0),
        ... (datetime(2016, 5, 9, 23, 59), 0.0)])
        >>> flat = flatten(kvs)
        >>> len(flat)
        2
        >>> flat[0]
        ('852-YF-008', datetime.datetime(2016, 5, 10, 0, 0), 0.0)
        """
        k, vs = kvs
        return [(k, v1, v2) for v1, v2 in vs]
    

In Python 2.7 you could also use lambda expression with tuple argument
unpacking but this is not portable and generally discouraged:

    
    
    lambda (k, vs): [(k, v1, v2) for v1, v2 in vs]
    

Version independent:

    
    
    lambda kvs: [(kvs[0], v1, v2) for v1, v2 in kvs[1]]
    

**Edit** :

If all you need is writing partitioned data then convert to Parquet directly
without `reduceByKey`:

    
    
    (sheet
        .flatMap(process)
        .map(lambda x: (x[0], ) + x[1])
        .toDF(["key", "datettime", "value"])
        .write
        .partitionBy("key")
        .parquet(output_path))
    

