는 어떻게 Kafka- 스파크 스트리밍 프로세스를 사용하여 메시지를 소비 할 응용 프로그램을 개발</p> <p>하나의 파일로 메시지 데이터 프레임을 스트리밍 카프카 - 불꽃을 저장하는 방법을 하나의 파일

Question

에 메시지 데이터 프레임을 스트리밍 카프카 - 불꽃을 저장합니다.

데이터가 수신되면 데이터 프레임으로 변환됩니다.

스트리밍 데이터 프레임이 텍스트 파일로 저장됩니다. 여기서 데이터 프레임은 각 카프카 스트림 메시지의 각 파일에 저장됩니다. 아래는 텍스트 프레임에 데이터 프레임으로 저장 한 코드입니다. 각 메시지에 대한 텍스트 파일을 mutiple합니다.

DF.coalesce(1).write.format("com.databricks.spark.csv").mode("append") 
           .save("path") 

여기에 내가 날짜 프레임을 스트리밍 달성하고자하는 요구 사항은 솔루션으로 저를 도와주세요 가능하면, 각 카프카 메시지에 대한 하나의 파일로 저장해야합니다.

미리 감사드립니다.

Answer 1

아래 코드가 도움이 될 수 있습니다. RDD 목록을 생성 한 다음이를 결합하십시오.

var dStreamRDDList = new ListBuffer[RDD[String]] 
dStream.foreachRDD(rdd => 
    { 
     dStreamRDDList += rdd 
    }) 
val joinRDD = ssc.sparkContext.union(dStreamRDDList) 
//then convert joinRDD to DataFrame (DF) 
DF.coalesce(1).write.format("com.databricks.spark.csv").mode("append") 
          .save("path")