你可以用一个映射函数来解决你的问题。请看以下代码:
df_new = spark.createDataFrame([
( 25,"Ankit","Ankit","Ankit"),( 22,"Jalfaizy","aa","Jalfaizy"),( 26,"aa","bb","Bala")
], ("age", "lname","mname","name"))
#only 3 records added to dataset
def find_identical(row):
labels = ["lname","mname","name"]
result = [row[0],] #save the age for final result
row = row[1:] #drop the age from row
for i in range(3):
s = []
field = row[i]
if field == row[(i+1)%3]: #check whether field is identical with next field
s.append(labels[(i+1)%3])
if field == row[(i-1)%3]: #check whether field is identical with previous field
s.append(labels[(i-1)%3])
if not s: #if no identical values found return None
s = None
result.append(s)
return result
df_new.rdd.map(find_identical).toDF(["age","lname_map_same","mname_map_same","name_map_same"]).show()
输出:
+---+--------------+--------------+--------------+
|age|lname_map_same|mname_map_same| name_map_same|
+---+--------------+--------------+--------------+
| 25| [mname, name]| [name, lname]|[lname, mname]|
| 22| [name]| null| [lname]|
| 26| null| null| null|
+---+--------------+--------------+--------------+
df_new = spark.createDataFrame([
( 25,"Ankit","Ankit","Ankit","Ankit","Ankit"),( 22,"Jalfaizy","aa","Jalfaizy","Jalfaizy","aa"),( 26,"aa","bb","Bala","cc","dd")
], ("age", "lname","mname","name","n1","n2"))
def find_identical(row):
labels = ["lname","mname","name","n1","n2"]
result = [row[0],]
row = row[1:]
for i in range(5):
s = []
field = row[i]
if field == row[(i+1)%5]:
s.append(labels[(i+1)%5])
if field == row[(i-1)%5]:
s.append(labels[(i-1)%5])
if field == row[(i+2)%5]:
s.append(labels[(i+2)%5])
if field == row[(i+3)%5]:
s.append(labels[(i+3)%5])
if not s:
s = None
result.append(s)
return result
df_new.rdd.map(find_identical).toDF(["age","lname_map_same","mname_map_same","name_map_same","n1_map_same","n2_map_same"]).show(truncate=False)
输出:
+---+---------------------+---------------------+----------------------+------------------------+------------------------+
|age|lname_map_same |mname_map_same |name_map_same |n1_map_same |n2_map_same |
+---+---------------------+---------------------+----------------------+------------------------+------------------------+
|25 |[mname, n2, name, n1]|[name, lname, n1, n2]|[n1, mname, n2, lname]|[n2, name, lname, mname]|[lname, n1, mname, name]|
|22 |[name, n1] |[n2] |[n1, lname] |[name, lname] |[mname] |
|26 |null |null |null |null |null |
+---+---------------------+---------------------+----------------------+------------------------+------------------------+
动态方法将列数作为参数。但是在我的例子中,这个数字应该在1到5之间,因为数据集是用最多5个属性创建的。可能是这样的:
df_new = spark.createDataFrame([
( 25,"Ankit","Ankit","Ankit","Ankit","Ankit"),( 22,"Jalfaizy","aa","Jalfaizy","Jalfaizy","aa"),( 26,"aa","bb","Bala","cc","dd")
], ("age", "n1","n2","n3","n4","n5"))
def find_identical(row,number):
labels = []
for n in range(1,number+1):
labels.append("n"+str(n)) #create labels dynamically
result = [row[0],]
row = row[1:]
for i in range(number):
s = []
field = row[i]
for x in range(1,number):
if field == row[(i+x)%number]:
s.append(labels[(i+x)%number]) #check for similarity in all the other fields
if not s:
s = None
result.append(s)
return result
number=4
colNames=["age",]
for x in range(1,number+1):
colNames.append("n"+str(x)+"_same") #create the 'nX_same' column names
df_new.rdd.map(lambda r: find_identical(r,number)).toDF(colNames).show(truncate=False)
根据number参数,输出会有所不同,我将age列作为第一列静态保留。
输出:
+---+------------+------------+------------+------------+
|age|n1_same |n2_same |n3_same |n4_same |
+---+------------+------------+------------+------------+
|25 |[n2, n3, n4]|[n3, n4, n1]|[n4, n1, n2]|[n1, n2, n3]|
|22 |[n3, n4] |null |[n4, n1] |[n1, n3] |
|26 |null |null |null |null |
+---+------------+------------+------------+------------+
