kafkaを通してストリーミングを開始するavroを送信する

Question

出力をバイト配列形式で取得していますが、文字列に変換する関数はありますか？

stream.foreachRDD(rdd => { 
    rdd.foreach(record => parseAVROToString(record)) 
}) 
stream.print 
ssc.start() 

message- 
{ 
    "id":"1", 
    "url":"http://localhost.com", 
    "article":" xyz", 
    "timestamp": 1366150681 
} 

受信： [B 1ffc17a0

@私はそれを文字列に変換したいです。

Answer 1

stream.foreachRDD((rdd, time) => { 
       val sqlContext = new SQLContext(sc) 
       import sqlContext.implicits._ 

       val jsonString = rdd.map { case (k, v) => AvroUtil.parseAVROToString(v) } 
       try { 
       val result = jsonString.mapPartitions(records => { 
        val mapper = new ObjectMapper() 
        mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false) 
        mapper.registerModule(DefaultScalaModule) 
        records.flatMap(record => { 
        try { 
         Some(mapper.readValue(record, classOf[Article])) 
        } catch { 
         case e: Exception => None 
        } 
        }) 
       }, true) 

       val df1 = result.toDF() 
       df1.show() 
       df1.write.format("json").mode(org.apache.spark.sql.SaveMode.Append).json("jsonresults") 
       //df1.save("org.apache.phoenix.spark", org.apache.spark.sql.SaveMode.Overwrite, Map("table" -> "articles".toUpperCase,"zkUrl" -> zkQuorum+":\hbase-unsecure")) 
       } catch { 
       case e: Exception => None; 
       } 
      }