七、Structured Streaming Kafka

朱雀 2024-04-18 16:01 67阅读 0赞

## 1、pom.xml ##

<properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <maven.compiler.compilerVersion>1.8</maven.compiler.compilerVersion>
        <scala.version>2.11.4</scala.version>
        <spark.version>2.4.1</spark.version>
      </properties>
    
      <repositories>
        <repository>
          <id>scala-tools.org</id>
          <name>Scala-Tools Maven2 Repository</name>
          <url>http://scala-tools.org/repo-releases</url>
        </repository>
      </repositories>
    
      <pluginRepositories>
        <pluginRepository>
          <id>scala-tools.org</id>
          <name>Scala-Tools Maven2 Repository</name>
          <url>http://scala-tools.org/repo-releases</url>
        </pluginRepository>
      </pluginRepositories>
    
      <dependencies>
        <dependency>
          <groupId>org.scala-lang</groupId>
          <artifactId>scala-library</artifactId>
          <version>${scala.version}</version>
        </dependency>
        
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
          <version>2.4.4</version>
          
        </dependency>
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-streaming_2.11</artifactId>
          <version>${spark.version}</version>
          <scope>${scope}</scope>
        </dependency>
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
          <version>${spark.version}</version>
        </dependency>
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-sql_2.11</artifactId>
          <version>${spark.version}</version>
          
        </dependency>
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-core_2.11</artifactId>
          <version>${spark.version}</version>
          
        </dependency>
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-sql_2.11</artifactId>
          <version>${spark.version}</version>
          
        </dependency>
    
        <dependency>
          <groupId>mysql</groupId>
          <artifactId>mysql-connector-java</artifactId>
          <version>5.1.34</version>
        </dependency>
      </dependencies>

## 2、Bean ##

创建所需要的Bean
    @JsonIgnoreProperties(ignoreUnknown = true)
    class DateBean extends Serializable {
      var item:Array[ItemBean] = Array.empty
      var sid = ""
      var roomid = ""
      var sourcelink=""
    }

## 3、工具类 ##

object KafkaStreamMapUtil {
      def kafkaStreamToDStream(df:DataFrame, chatTypeName:String):
            RDD[Array[(Array[MessageBean],Array[GiftBean])]] = {
        df.rdd.map(row=>{
          val jsonParse = new JsonParser()
          val gs =new Gson()
          var msg:scala.Array[(Array[MessageBean],Array[GiftBean])]=null
          val  sdf = new SimpleDateFormat("yyyyMMdd")
          try {
            //        println(line._2)
            val je=jsonParse.parse(row.getAs[String]("value"))
            var dbArr:Array[DateBean]=null
            if(je.isJsonArray){
              dbArr= gs.fromJson(je,classOf[Array[DateBean]])
            }else{
              val dbb= gs.fromJson(je,classOf[DateBean])
              dbArr= new Array[DateBean](1)
              dbArr(0)=dbb
            }
            msg = dbArr.map(m=>{
              var p=""
              var r=""
              var mArr:Array[MBean]=Array.empty
              var gArr:Array[GBean]=Array.empty
              if(m!=null & m.item!=null & m.item.length>0){
                p=m.sid
                r=m.r
                var gBean :GBean= null
                var mBean :MBean = null
                m.item.foreach(x=> {
                  gBean = new GBean
                  mBean = new MBean
                  if(x!=null){
                    if(x.typeName.equals(chatTypeName)){
                      mBean .p =p
                      mBean .r =r
                      mBean .f = x.f
                      mBean .timestamp =x.time
                      mBean .content=x.content
                      val date  = sdf.format(new Date(x.time.toLong * 1000))
                      mBean .date = date
                      mArr :+= mBean 
                    } else{
                      if(!x.price.trim.equals("")){
                        //                    println(line._2)
                        gBean .p = p
                        gBean .r =r
                        gBean .f = x.f
                        gBean .fname = x.fname
    
                        gBean .g_type= x.g_type
                        gBean .price = x.price.toDouble
                        gBean .count = x.count
                        gBean .timestamp = x.time
                        gBean .g_name=x.gname
                        gBean .g_id=x.gid
                        val date  = sdf.format(new Date(x.time.toLong * 1000))
                        gBean .date = date
                        gArr :+= gBean 
                      }
                    }
                  }
                })
              }
              (msgArr,giftArr)
            })
          } catch {
            case e: Exception =>println(e.getMessage)
          }
          msg
        })
    
      }
    }

## 4、主类 ##

object KafkaSourceApp {
      def main(args: Array[String]): Unit = {
        Logger.getLogger("org").setLevel(Level.ERROR)
        val sqLContext = SparkSession.builder().appName(" structured_streaming_kafka_App")
            .master("local[*]").getOrCreate()
        val df = sqLContext
          .readStream
          .format("kafka")
          .option("kafka.bootstrap.servers", "brokerid")
          .option("subscribe", "topic")
          .load()
        val strDF = df
          .selectExpr( "CAST(value AS STRING)")
        val chatTypeName="chat"
        strDF.writeStream.format("console")
          .option("checkpointLocation", "path/")
          .foreachBatch((df,id)=>{
            println(s"batchid = $id")
            import sqLContext.implicits._
            val transRDD  = KafkaStreamMapUtil.kafkaStreamToDStream(df,chatTypeName)
            val msgDF = transRDD.flatMap(x=>x)
              .flatMap(_._1)
              .map(e=>(e.p,e.r,e.f,e.content,e.timestamp,e.date))
              .toDF("p","r","f","content","timestamp","date")
              .selectExpr("*","platform_id as plat")
            val giftDF =transRDD.flatMap(x=>x)
              .flatMap(_._2)
              .map(e=>(e.p,e.r,e.f,e.f_name,e.g_type,e.price,e.count,e.timestamp,e.date,e.g_name,e.g_id))
              .toDF("p","r","f","fname","g_type",
                        "price","count","timestamp","date","g_name","g_id")
              .selectExpr("*","platform_id as plat")
            msgDF.show()
            giftDF.show()
          }).start().awaitTermination()
      }
    }