Spark'da Spark DataFrame'i Oluşturun JSON'dan Mesaj Aktarılıyor

Question

Scala'da Spark Streaming uygulamasının bir kafka konusundan JSON Strings'i alıp bir veri çerçevesine yüklemek istediğim bir uygulama üzerinde çalışıyorum. Spark'in şemaya kendi başına bir RDD'den (String) ulaştığı bir yol var mı?

sqlContext.read 
//.schema(schema) //optional, makes it a bit faster, if you've processed it before you can get the schema using df.schema 
.json(jsonRDD) //RDD[String] 

ben yapmaya çalışıyorum:

val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc) 
    val yourDataFrame = hiveContext.createDataFrame(yourRDD) 

Answer 1

kıvılcım 1.4 olarak, RDD bir Dataframe oluşturmak için aşağıdaki yöntemi deneyebilirsiniz aynı anda. Yine de, Kafka'nın dışındaki RDD [String] 'inin nasıl olduğunu merak ediyorum, hala Spark + Kafka'nın sadece "şu anda orada olanı çıkar" yerine tek seferlik bir yayın olduğu izlenimi altındayım. :)

Answer 2

Evet, aşağıdakileri kullanabilirsiniz:

Answer 3

JSON değerlerini çıkarmak, Kafka gelen mesajların akışında okumak için aşağıdaki kodu kullanabilirsiniz ve DataFrame çevirebiliriz:

val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet) 

messages.foreachRDD { rdd => 
//extracting the values only 
    val df = sqlContext.read.json(rdd.map(x => x._2)) 
    df.show() 
}