spark：ReceiverLauncher.launchによって作成されたDSStreamであるforeachRDDによってデータを取得する方法は？

Question

// consumer data from kafka 

val tmp_stream = ReceiverLauncher.launch(ssc, props, numberOfReceivers,StorageLevel.MEMORY_ONLY) 

tmp_stream.foreachRDD(rdd => { 
    rdd.collect() 
    val count = rdd.count() // here I can get the count of datas 
    // How can I get data here ? 
}) 

ReceiverLauncher.launch

Answer 1

によって作成されたストリームからforeachRDDでデータを取得するには、このコードを完了するためにどのように任意のアイデアあなたは、生のbyte[]を取得するためにgetPayLoadを呼び出すことができます。このようなもの

val stream = tmp_stream.map(x => { val s = new String(x.getPayload); s })