df_schema = StructType([StructField('p_id', StringType(), True),
                        StructField('c_id_map', MapType(StringType(), StringType(), True), True),
                        StructField('d_id', LongType(), True)])

df = sqlContext.createDataFrame(hour_filtered_rdd, df_schema)
dfwriter = df.write
dfwriter.mode('overwrite')
dfwriter.format('parquet')

c_id_map: