如何在我的 pyspark 代码中解决这个 reducebykey 转换问题？答案

【问题标题】：How to fix this reducebykey transformation issue in my pyspark code?如何在我的 pyspark 代码中解决这个 reducebykey 转换问题？
【发布时间】：2019-02-06 16:05:38
【问题描述】：

对于如何正确获取此值，我有点困惑。以下是我的示例数据：

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

我正在尝试将输出数据框设为：

我可以到此为止。在这里需要你的帮助。

    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)

我当前的代码输出：

它没有根据段 id 和 CatColName 正确分组数据。

【问题讨论】：

名称必须是segment 1吗？或者是否可以在那里添加其他值
是的，这是必要的，因为在创建数据框后，我计划生成一个 json

标签： apache-spark pyspark apache-spark-sql pyspark-sql

【解决方案1】：

问题是 reduceByKey 将您生成的字符串 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 只需要恢复原名即可。

【讨论】：