apache-spark - 如何在我的 pyspark 代码中解决这个 reducebykey 转换问题?
问题描述
我对如何正确获取这个值有点困惑。以下是我的示例数据:
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 正确分组数据。
解决方案
问题是 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 只需要恢复原始名称。
推荐阅读
- javascript - 边缘没有在画布上正确设置高度
- groovy - 如何使用 groovy 脚本引擎替换三元运算中的变量
- java - 通配符类作为构造函数中的变量
- javascript - 两者都具有默认文件输入的 Dropzone
- ruby-on-rails - 下载 DataTables 的最简单方法是什么?
- c++11 - 没有 #define 的有效 C 预处理器宏
- security - 如何针对 OWASP Benchmark 正确运行 ZAP?
- swift - rx-swift 重新加载表格后在 UITableView 中保持选择
- node.js - Heroku 配置变量在 process.env 上不可用
- scala - 如何使用 Scala 将数据作为元组传递到 Spark 中的 rdd