Pandas Grouped Map UserDefinedFunction mixes column labels

concat_hours

#!/usr/bin/env python3
# coding: utf-8

input_data = b"""en Albert_Camus 20071210-000000 150
en Albert_Camus 20071210-010000 148
en Albert_Camus 20071210-020000 197
en Albert_Camus 20071211-200000 145
en Albert_Camus 20071211-210000 131
en Albert_Camus 20071211-220000 154
en Albert_Camus 20071211-230001 142
en Albert_Caquot 20071210-020000 1
en Albert_Caquot 20071210-020000 1
en Albert_Caquot 20071210-040001 1
en Albert_Caquot 20071211-060000 1
en Albert_Caquot 20071211-080000 1
en Albert_Caquot 20071211-150000 3
en Albert_Caquot 20071211-210000 1"""

import tempfile

fp = tempfile.NamedTemporaryFile()
fp.write(input_data)

fp.seek(0)

import findspark
findspark.init()

import pyspark
from pyspark.sql.types import StructType, StructField
from pyspark.sql.types import StringType, IntegerType, TimestampType
from pyspark.sql import functions

sc = pyspark.SparkContext(appName="udf_example")
sqlctx = pyspark.SQLContext(sc)

schema = StructType([StructField("lang", StringType(), False),
                     StructField("page", StringType(), False),
                     StructField("timestamp", TimestampType(), False),
                     StructField("views", IntegerType(), False)])

df = sqlctx.read.csv(fp.name,
                     header=False,
                     schema=schema,
                     timestampFormat="yyyyMMdd-HHmmss",
                     sep=' ')

df.count()
df.dtypes
df.show()

new_schema = StructType([StructField("lang", StringType(), False),
                         StructField("page", StringType(), False),
                         StructField("day", StringType(), False),
                         StructField("enc", StringType(), False)])

from pyspark.sql.functions import pandas_udf, PandasUDFType
import pandas as pd

hour_to_letter = ['A','B','C','D','E','F','G','H','I','J','K','L','M','N','O',
                  'P','Q','R','S','T','U','V','W','X']

@pandas_udf(new_schema, PandasUDFType.GROUPED_MAP)
def concat_hours(x):
    view_hours = x['hour'].tolist()
    view_views = x['views'].tolist()

    view_hours_letters = [hour_to_letter[h] for h in view_hours]

    encoded_views = [l + str(h)
                     for l, h in sorted(zip(view_hours_letters,view_views))]
    encoded_views_string = ''.join(encoded_views)

    # return pd.DataFrame({'enc': x.page, 'day': x.lang, 'lang': x.day,
    #                      'page': encoded_views_string}, index=[x.index[0]])

    return pd.DataFrame({'page': x.page, 'lang': x.lang,'day': x.day,
                         'enc': encoded_views_string}, index=[x.index[0]])


from pyspark.sql import functions
grouped_df = (df.select(['lang',
                         'page',
                         functions.date_format('timestamp','yyyy-MM-dd')\
                                  .alias('day'), 
                         functions.hour('timestamp').alias('hour'), 
                         'views'
                         ])
                .groupby(['lang','page','day'])
                )

grouped_df = (grouped_df.apply(concat_hours)
                        .dropDuplicates()
                        )

grouped_df.show()