希望这可以帮助！
（ 编辑说明： 在获取更新的需求之后对代码进行了调整）
import pyspark.sql.functions as func#create RDDrdd = sc.parallelize([(22,'C','xxx','yyy','zzz'),(23,'D','xxx','yyy','zzz'),(24,'C','xxx1','yyy1','zzz1'),(25,'D','xxx1','yyy1','zzz1')])#convert RDD to dataframedf = rdd.toDF(['Date','Type','Data1','Data2','Data3'])df.show()#group by 3 data columns to create list of date & typedf1 = df.sort("Data1","Data2","Data3","Type").groupBy("Data1","Data2","Data3").agg(func.collect_list("Type"),func.collect_list("Date")).withColumnRenamed("collect_list(Type)", "Type_list").withColumnRenamed("collect_list(Date)", "Date_list")#add 2 new columns by splitting above date list based on type list's valuedf2 = df1.where((func.col("Type_list")[0]=='C') & (func.col("Type_list")[1]=='D')).withColumn("Start Date",df1.Date_list[0]).withColumn("End Date",df1.Date_list[1])#select only relevant columns as an outputdf2.select("Data1","Data2","Data3","Start Date","End Date").show()另一种解决方案采用RDD： -
（ 编辑注： 以下片段添加@AnmolDave感兴趣RDD解决方案，以及）
import pyspark.sql.types as typrdd = sc.parallelize([('xxx','yyy','zzz','C',22),('xxx','yyy','zzz','D',23),('xxx1','yyy1','zzz1','C', 24),('xxx1','yyy1','zzz1','D', 25)])reduced = rdd.map(lambda row: ((row[0], row[1], row[2]), [(row[3], row[4])])) .reduceByKey(lambda x,y: x+y) .map(lambda row: (row[0], sorted(row[1], key=lambda text: text[0]))) .map(lambda row: ( row[0][0], row[0][1], row[0][2], ','.join([str(e[0]) for e in row[1]]), row[1][0][1], row[1][1][1] ) ) .filter(lambda row: row[3]=="C,D")schema_red = typ.StructType([ typ.StructField('Data1', typ.StringType(), False), typ.StructField('Data2', typ.StringType(), False), typ.StructField('Data3', typ.StringType(), False), typ.StructField('Type', typ.StringType(), False), typ.StructField('Start Date', typ.StringType(), False), typ.StructField('End Date', typ.StringType(), False) ])df_red = sqlContext.createDataframe(reduced, schema_red)df_red.show()
