小编Kal*_*hal的帖子

我有一个数据名气，看起来像这样

Sno|UserID|TypeExp
1|JAS123|MOVIE
2|ASP123|GAMES
3|JAS123|CLOTHING
4|DPS123|MOVIE
5|DPS123|CLOTHING
6|ASP123|MEDICAL
7|JAS123|OTH
8|POQ133|MEDICAL
.......
10000|DPS123|OTH

UserID 是标识用户的列，TypeExp 列定义其该月的支出类型，现在我可能有 5 种不同的支出，即

TypeExpList=[电影、游戏、服装、医疗、其他]

现在我想将其转换为用户级数据帧，其中有一个 0 或 1 二进制变量存储信息天气或用户“X”是否已完成上述支出类型

例如，在上面的快照中，DataFrame 输出应该如下所示

User| TypeExpList         #Type list is this array corresponding entry's [MOVIE,GAMES,CLOTHING,MEDICAL,OTH]
JAS123 |[1,0,1,0,1]      #since user has done expenditure on Movie,CLOTHING,OTHER Category 
ASP123 |[0,1,0,1,0]       #since User expenditure on  GAMES & MEDICAL
DPS123 |[1,0,1,0,1]         #since user expenditure on  MOVIE,CLOTHING & OTHER
POQ133 |[0,0,0,1,0]        #since User Expenditure on MEDICAL only 

我在 Spark 中有一个数据框，看起来像

事件DF

   Sno|UserID|TypeExp
    1|JAS123|MOVIE
    2|ASP123|GAMES
    3|JAS123|CLOTHING
    4|DPS123|MOVIE
    5|DPS123|CLOTHING
    6|ASP123|MEDICAL
    7|JAS123|OTH
    8|POQ133|MEDICAL
    .......
    10000|DPS123|OTH

我需要以 Avro 格式将其写入 Kafka 主题，目前我可以使用以下代码在 Kafka 中将其写入为 JSON

val kafkaUserDF: DataFrame = eventDF.select(to_json(struct(eventDF.columns.map(column):_*)).alias("value"))
  kafkaUserDF.selectExpr("CAST(value AS STRING)").write.format("kafka")
    .option("kafka.bootstrap.servers", "Host:port")
    .option("topic", "eventdf")
    .save()

现在我想以 Avro 格式将其写入 Kafka 主题

我有一个看起来像的数据框

数据

*id*,             *name*,                      *URL*,                 *Type*  
    2,             birth_france_by_region,    http://abc. com,       T1 
    2,             birth_france_by_region,    http://pt. python,     T2 
    3,             long_lat,                  http://abc. com,       T3 
    3,             long_lat,                  http://pqur. com,      T1 
    4,             random_time_series,        http://sadsdc. com,    T2 
    4,             random_time_series,        http://sadcadf. com,   T3
    5,             birth_names,               http://google. com,    T1 
    5,             birth_names,               http://helloworld. com,T2 
    5,             birth_names,               http://hu. com,        T3

我想要一个此数据帧合并id相等的行，并具有对应于URL的Type列表， 所以最终输出应类似于

*id*, *name*,             *URL*,                               *Type*  
2,birth_france_by_region,  [http://abc .com,http://pt.python], [T1,T2] 
3,long_lat,           [http://abc .com,http://pqur. com],       [T3,T1] 
4,random_time_series, [http://sadsdc. com,http://sadcadf .com,],[T2,T3] 
5,birth_names,        [http://google .com,http://helloworld. com,
                                       http://hu. com] , …