Как исправить эту проблему с преобразованием limitbykey в моем коде pyspark? - PullRequest
Новая
       1

Как исправить эту проблему с преобразованием limitbykey в моем коде pyspark?

0 голосов
/ 06 февраля 2019

Я немного застрял, как получить это значение правильно.Ниже приведены мои образцы данных: 

col_name,Category,SegmentID,total_cnt,PercentDistribution
city,ANTIOCH,1,1,15
city,ARROYO GRANDE,1,1,15
state,CA,1,3,15
state,NZ,1,4,15

enter image description here

Я пытаюсь получить выходной кадр данных как:

enter image description here

Я мог бы приехать до этого.Здесь вам нужна помощь. 

    from pyspark.sql.types import StructType,StructField,StringType,IntegerType
    import json

    join_df=spark.read.csv("/tmp/testreduce.csv",inferSchema=True, header=True)
    jsonSchema = StructType([StructField("Name", StringType())
                           , StructField("Value", IntegerType())
                           , StructField("CatColName", StringType())
                           , StructField("CatColVal", StringType())
                        ])
    def reduceKeys(row1, row2):
            row1[0].update(row2[0])
            return row1

    res_df=join_df.rdd.map(lambda row: ("Segment " + str(row[2]), ({row[1]: row[3]},row[0],row[4])))\
.reduceByKey(lambda x, y: reduceKeys(x, y))\
.map(lambda row: (row[0], row[1][2],row[1][1], json.dumps(row[1][0]))).toDF(jsonSchema)

Мой текущий код вывода:

Неправильно группируются данные на основе идентификатора сегмента и CatColName.

enter image description here

1 Ответ

0 голосов
/ 06 февраля 2019

Проблема состоит в том, что reduByKey учитывает сгенерированную строку Segment 1, и это равно для города и штата.Если вы добавите col_name в начале, он будет работать, как и ожидалось, но вы получите другое имя в своем результате.Это можно изменить с помощью регулярного выражения 

res_df=test_df.rdd.map(lambda row: ("Segment " + str(row[2]) +" " + str(row[0]), ({row[1]: row[3]},row[0],row[4])))\
.reduceByKey(lambda x, y: reduceKeys(x, y))\
.map(lambda row: (row[0], row[1][2],row[1][1], json.dumps(row[1][0]))).toDF(jsonSchema).withColumn("name",regexp_extract(col("name"),"(\w+\s\d+)",1))

res_df.show(truncate=False)

Вывод: 

+---------+-----+----------+----------------------------------+
|name     |Value|CatColName|CatColVal                         |
+---------+-----+----------+----------------------------------+
|Segment 1|15   |city      |{"ANTIOCH": 1, "ARROYO GRANDE": 1}|
|Segment 1|15   |state     |{"CA": 3, "NZ": 4}                |
+---------+-----+----------+----------------------------------+

Окончательный regexp_extract необходим только для восстановления исходного имени.

...