python之Spark 流 : read CSV string from kafka, 写入 Parquet

有很多从 Kafka 读取 json(写入 parquet)的在线示例 - 但我不知道如何将模式应用于来自 kafka 的 CSV 字符串。

流式数据:

customer_1945,cusaccid_995,27999941     
customer_1459,cusaccid_1102,27999942 

架构:

schema = StructType() \ 
.add("customer_id",StringType()) \ 
.add("customer_acct_id",StringType()) \ 
.add("serv_acct_id",StringType()) 

读取流:

df = spark \ 
  .readStream \ 
  .format("kafka") \ 
  .option("kafka.bootstrap.servers", "xx.xx.xx.xx:9092") \ 
  .option("subscribe", "test") \ 
  .load() 

我将其用于 JSON:

interval=df \ 
  .select(from_json(col("value").cast("string"), schema).alias("json")) \ 
  .select("json.*") 

在使用分配的模式将其写入 parquet 之前:

query=interval     \ 
  .writeStream  \ 
  .format("parquet") \ 
  .option("checkpointLocation", "/user/whatever/checkpoint24") \ 
  .start("/user/ehatever/interval24") 

因为我不能将 from_json() 用于 CSV - 我不知道如何将模式应用到数据帧以便我可以使用类似的 writeStream() 命令。

请您参考如下方法:

我是这样做的。没有 from_json，提取 csv 字符串:

interval=df.select(col("value").cast("string")) .alias("csv").select("csv.*") 

然后将其拆分为列。这可以使用上面相同的语句写成 Parquet 文件

interval2=interval \ 
      .selectExpr("split(value,',')[0] as customer_id" \ 
                 ,"split(value,',')[1] as customer_acct_id" \ 
                 ,"split(value,',')[2] as serv_acct_id" \ 
                 ,"split(value,',')[3] as installed_service_id" \ 
                 ,"split(value,',')[4] as meter_id" \ 
                 ,"split(value,',')[5] as channel_number" \ 
                 ... etc 
                 )