pyspark RDD 一行转多行

dataframe也有按某列中按分隔符转成多行的函数，不过dataframe比rdd所需资源更多，所以此处先些rdd的按某列拆分转为多行
dataframe的详见https://spark.apache.org/docs/2.3.1/api/python/pyspark.sql.html#pyspark.sql.DataFrame 的pyspark.sql.functions.explode(col) 、pyspark.sql.functions.explode_outer(col) 、pyspark.sql.functions.posexplode(col)、pyspark.sql.functions.posexplode_outer(col)函数接口split切割方法

rdd的拆分主要使用https://spark.apache.org/docs/2.3.1/api/python/pyspark.html#pyspark.RDD 的flatMap(f, preservesPartitioning=False)]、flatMapValues(f)方法

flatMap 按传输的参数和调用的f函数来拆分

x = sc.parallelize([("a", "x,y,z"), ("b", "p,r")])
x.flatMap(lambda x:[(x[0],x[1].split(",")[0]),(x[0],x[1].split(",")[1])]).collect() #取第二列的按‘，’号分隔前两个作为两行，但是对未知个数不太适用,而且实际生产因为数据不标准容易致使程序异常
[('a', 'x'), ('a', 'y'), ('b', 'p'), ('b', 'r')]

def itemsToRow(x):
	    list=[]
	    for value in x[1].split(","):
		    newrow=(x[0],value)
		    list.append(newrow)
		    return list
x.flatMap(lambda x:itemsToRow(x)).collect()#这样就将未知个数的有多少个元素就弄成多少行了

flatMapValues就比较简单，就是单纯的将一行的(k,v)中的v传进去的v容器元素进行拆分，##### 注意v是容器类型

x = sc.parallelize([("a", ["x", "y", "z"]), ("b", ["p", "r"])])
def f(x): return x
x.flatMapValues(f).collect()
[('a', 'x'), ('a', 'y'), ('a', 'z'), ('b', 'p'), ('b', 'r')]

如果v不是容器类型，就需要想棒法弄成容器类型，例如:

x = sc.parallelize([("a", "x,y,z"), ("b", "p,r")])
def f(x): return x
x.map(lambda x:(x[0],x[1].split(","))).flatMapValues(f).collect()
[('a', 'x'), ('a', 'y'), ('a', 'z'), ('b', 'p'), ('b', 'r')]

此方法与combineByKey将多行弄成一行有相反的功能机制，结合起来反复玩耍，可玩出许多花样，堪称强强组合，你值得拥有

二、实战
原数据格式如下

现在需将以上表格转换成以下格式：

def f(x): return x

t1.rdd.map(lambda x: ((x[0], x[1]), json.loads(x[2]))).flatMapValues(f)
        .map(lambda x: (x[0][0], x[0][1], json.dumps(x[1][0])))
        .toDF(['id','contentable_id','material_id'])
  

坑：需将原C列，从array转成list格式，否则在toDF()操作时，字典的value值会变成None，

最后

以上就是安详早晨为你收集整理的pyspark RDD 一行转多行的全部内容，希望文章能够帮你解决pyspark RDD 一行转多行所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。