from_ json не преобразует json в DF в искровой структурированной потоковой передаче - PullRequest
Новая
       106

from_ json не преобразует json в DF в искровой структурированной потоковой передаче

0 голосов
/ 18 июня 2020

Привет, я новичок в Spark Streaming. Я пытаюсь реализовать потоковое решение, которое будет читать json сообщение от kafka и сохранять его в Cassandra. Проблема, с которой я столкнулся, - from_ json не преобразовывает json в класс Case.

Вот мой Json: 

{"brand":"hortense","category":"clothing","description":"Hortense B. Hewitt 25027 Peplum Garter Set","mode":"dinner's club","orditems":"2","productcode":"8f6e9f55-c69d-4b2c-a249-572b4e53fa9a","unitprice":"3360"}

build.sbt 

scalaVersion := "2.11.8"
val spark="2.3.1"
val kafka="0.10.1"
val cassandra="3.2"
val cassandraConnectot="2.3.0"


// https://mvnrepository.com/artifact/org.apache.kafka/kafka
//Tips Taken from:https://www.scala-sbt.org/1.x/docs/Resolvers.html
resolvers += "DefaultMavenRepository" at "https://mvnrepository.com/artifact/"
dependencyOverrides += "com.google.guava" % "guava" % "15.0"

dependencyOverrides += "com.fasterxml.jackson.core" % "jackson-core" % "2.9.6"
dependencyOverrides += "com.fasterxml.jackson.core" % "jackson-databind" % "2.9.6"
dependencyOverrides += "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.9.6"
// https://mvnrepository.com/artifact/com.fasterxml.jackson.dataformat/jackson-dataformat-cbor
dependencyOverrides += "com.fasterxml.jackson.dataformat" % "jackson-dataformat-cbor" % "2.9.6"

//libraryDependencies += "org.scala-sbt" % "sbt" % "1.2.8" % "provided"
libraryDependencies += "org.apache.spark" % "spark-streaming_2.11" % spark
libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka-0-10" %spark
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.3.1"
libraryDependencies +="com.typesafe.play" %"play-json_2.11" % "2.5.0" exclude("com.fasterxml.jackson.core","jackson-databind")
libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.3.1"
libraryDependencies +="com.typesafe" % "config" %"1.3.2"
libraryDependencies +="com.datastax.spark" %% "spark-cassandra-connector" % cassandraConnectot
libraryDependencies +="com.datastax.spark" %% "spark-cassandra-connector-embedded" % cassandraConnectot % Test
libraryDependencies += "org.apache.spark" %% "spark-sql-kafka-0-10" % "2.3.1" % "provided"
libraryDependencies += "com.datastax.spark" % "spark-cassandra-connector_2.11" % "2.3.0"

========== ===================== Главный класс =========================== =========================== 

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions.from_json
import org.apache.spark.sql.streaming.Trigger
//spark-submit --master local --driver-memory 1g --executor-memory  1g   --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0  --class
// TransctionReceiverStructuredTreaming   /Users/abhinav/Downloads/SparkStreamingExample/target/scala-2.11/sparkstreamingexample_2.11-0.1.0-SNAPSHOT.jar

object TransctionReceiverStructuredTreaming  extends  SparkSessionBuilder {
  def main(args: Array[String]) {
    case class TransctionData(productcode: String, description: String, brand: String, category: String, unitprice: String, orditems: String, mode: String)
    val transactionSchema = StructType(Array(
      StructField("brand", StringType, true),
      StructField("category", StringType, true),
      StructField("description", StringType, true),
      StructField("mode", StringType, true),
      StructField("orditems", DoubleType, true),
      StructField("productcode", StringType, true),
      StructField("unitprice", StringType, true)))


    val spark = buildSparkSession
    import spark.implicits._
    /*val spark = SparkSession.builder
    .master("local")
    .appName("TransctionReceiver")
    .getOrCreate();*/
    val ssc = new StreamingContext(spark.sparkContext, Seconds(30))
    import spark.implicits._
    val topics = List("-first_topic")


    val rawKafkaDF = spark.sqlContext.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("subscribe", "-first_topic")
      .option("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
      .option("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
      .option("startingOffsets", "earliest")
   //   .option("endingOffsets", "latest")
      //.option("auto.offset.reset", "earliest")
     // .option("group.id", "group66554")
      .load()
//    println("rawKafkaDF writing in kafka>>>>"+rawKafkaDF.count())


    import spark.implicits._
    val df = rawKafkaDF
      .selectExpr("CAST(value AS STRING)").as[String]
      .flatMap(_.split("\n"))

  df.writeStream
      .format("console")
      .trigger(Trigger.Once())
      .start().awaitTermination()
    //val jsons = df.select(from_json($"value", transactionSchema) as "data").select("data.*")
    val jsons1 = df.select($"value" cast "string" as "json")
      .select(from_json($"json", transactionSchema) as "data")
      .select("data.*")

    jsons1.writeStream
      .format("console")
      .trigger(Trigger.Once())
      .start().awaitTermination()
    println("Print 2 end >>>>")
    val sink = jsons1
      .writeStream
      .queryName("KafkaToCassandraForeach")
      .outputMode("update")
      .foreach(new CassandraSinkForeach())
      .start()
    sink.awaitTermination()


    ssc.start()

  }
}

============== =============================== 

When I run this program:
I can see:
 df.writeStream
      .format("console")
      .trigger(Trigger.Once())
      .start().awaitTermination()
=====Giving O/P=====
+--------------------+
|               value|
+--------------------+
|{"brand":"adult",...|
|{"brand":"polo","...|
|{"brand":"timberl...|

+--------------------+

But from_json is not printing any data:Also in cassandra only Null has been entered.
  jsons1.writeStream
      .format("console")
      .trigger(Trigger.Once())
      .start().awaitTermination()

+-----+--------+-----------+----+--------+-----------+---------+
|brand|category|description|mode|orditems|productcode|unitprice|
+-----+--------+-----------+----+--------+-----------+---------+
| null|    null|       null|null|    null|       null|     null|
| null|    null|       null|null|    null|       null|     null|
| null|    null|       null|null|    null|       null|     null|
| null|    null|       null|null|    null|       null|     null|

Ребят Рабочее решение: 

import org.apache.spark.sql.{Dataset, Encoders, SparkSession}
import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions.{col, from_json}
import org.apache.spark.sql.streaming.Trigger
import sampleTestClass.Bean44
//spark-submit --master local --driver-memory 1g --executor-memory  1g   --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0  --class
// TransctionReceiverStructuredTreaming   /Users/abhinav/Downloads/SparkStreamingExample/target/scala-2.11/sparkstreamingexample_2.11-0.1.0-SNAPSHOT.jar
case class Bean44(brand:String,category:String,description:String,mode:String,orditems:String,productcode:String,unitprice:String)
object TransctionReceiverStructuredTreaming  extends  SparkSessionBuilder {
  def main(args: Array[String]) {
    case class TransctionData(productcode: String, description: String, brand: String, category: String, unitprice: String, orditems: String, mode: String)
    val transactionSchema = StructType(Array(
      StructField("brand", StringType, true),
      StructField("category", StringType, true),
      StructField("description", StringType, true),
      StructField("mode", StringType, true),
      StructField("orditems", DoubleType, true),
      StructField("productcode", StringType, true),
      StructField("unitprice", StringType, true)))


    val spark = buildSparkSession
    import spark.implicits._
    val ssc = new StreamingContext(spark.sparkContext, Seconds(30))

    val topics = List("-first_topic")
    val rawKafkaDF = spark.sqlContext.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("subscribe", "-first_topic")
      .option("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
      .option("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
      .option("startingOffsets", "earliest")
      .load()


    val schema = Encoders.product[Bean44].schema
    val df1 = rawKafkaDF
      .selectExpr("CAST(value AS STRING)").as[String]
      .flatMap(_.split("\n")).toDF()
    val df = df1.withColumn("data",from_json(col("value"),schema)).select("data.*").as[Bean44]


  df.writeStream
      .format("console")
      .trigger(Trigger.Once())
      .start().awaitTermination()

    val sink = df
      .writeStream
      .queryName("KafkaToCassandraForeach")
      .outputMode("update")
      .foreach(new CassandraSinkForeach())
      .start().awaitTermination()
    ssc.start()

  }

1 Ответ

1 голос
/ 18 июня 2020

Я думаю, вы были почти близки

Шаги, которые я выполнил

  1. Загрузить JSON как список строк в Dataframe
  2. Создано и кодировалось из Bean44
  3. Разобрал json, используя from_json, до Bean44 типа структуры
  4. df.select("data.*") так же, как вы, и добавил .as[Bean44], чтобы получить Dataset[Bean44]
import org.apache.spark.sql.{Dataset, Encoders}
import org.apache.spark.sql.functions._

object JsonToCase {

  def main(args: Array[String]): Unit = {

    val spark = Constant.getSparkSess

    import spark.implicits._

    val schema = Encoders.product[Bean44].schema

    var df = List("""{"brand":"hortense","category":"clothing","description":"Hortense B. Hewitt 25027 Peplum Garter Set","mode":"dinner's club","orditems":"2","productcode":"8f6e9f55-c69d-4b2c-a249-572b4e53fa9a","unitprice":"3360"}""").toDF("value")
    df = df.withColumn("data",from_json(col("value"),schema))
    val mappedTOBean: Dataset[Bean44] = df.select("data.*").as[Bean44]
    mappedTOBean.show()

  }

}

case class Bean44(brand:String,category:String,description:String,mode:String,orditems:String,productcode:String,unitprice:String)

Кодеры используются для преобразования объекта JVM типа T во внутреннее представление Spark SQL и обратно. Кодеры обычно создаются автоматически с помощью имплицитов из SparkSession или могут быть созданы явно путем вызова методов stati c на кодировщиках.

Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.

Похожие темы

...