1. 全栈程序员必看首页

Spark Streaming Join

全栈程序员-站长 • • 未分类 • 阅读 19

Spark Streaming Join多数据源Join思路多数据源Join大致有以下三种思路：数据源端Join，如Android/IOS客户端在上报用户行为数据时就获取并带上用户基础信息。计算引擎上Join,如用SparkStreaming、Flink做Join。结果端Join，如用HBase/ES做Join,Join键做Rowkey/_id，各字段分别写入列簇、列或field。三种思路各有优劣，使用时注意…

大家好，又见面了，我是你们的朋友全栈君。

多数据源Join思路

多数据源Join大致有以下三种思路：

  • 数据源端Join，如Android/IOS客户端在上报用户行为数据时就获取并带上用户基础信息。

  • 计算引擎上Join,如用Spark Streaming、Flink做Join。

  • 结果端Join，如用HBase/ES做Join,Join键做Rowkey/_id，各字段分别写入列簇、列或field。

三种思路各有优劣，使用时注意一下。这里总结在计算引擎Spark Streaming上做Join。

Stream-Static Join

流与完全静态数据Join

流与完全静态数据Join。有两种方式，一种是RDD Join方式，另一种是Broadcast Join(也叫Map-Side Join)方式。

RDD Join 方式

思路：RDD Join RDD 。

package com.bigData.spark

import com.alibaba.fastjson.{ 
   JSON, JSONException, JSONObject}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{ 
   Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{ 
   ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{ 
   Durations, StreamingContext}

/** * Author: Wang Pei * License: Copyright(c) Pei.Wang * Summary: * * Stream-Static Join * * spark 2.2.2 * */
case class UserInfo(userID:String,userName:String,userAddress:String)
object StreamStaicJoin { 
   
  def main(args: Array[String]): Unit = { 
   

    //设置日志等级
    Logger.getLogger("org").setLevel(Level.WARN)

    //Kafka 参数
    val kafkaParams= Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (true: java.lang.Boolean),
      "group.id" -> "testTopic3_consumer_v1")

    //spark环境
    val sparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.replace("$","")).setMaster("local[3]")
    val ssc = new StreamingContext(sparkConf,Durations.seconds(10))

    /** 1) 静态数据: 用户基础信息*/
    val userInfo=ssc.sparkContext.parallelize(Array(
      UserInfo("user_1","name_1","address_1"),
      UserInfo("user_2","name_2","address_2"),
      UserInfo("user_3","name_3","address_3"),
      UserInfo("user_4","name_4","address_4"),
      UserInfo("user_5","name_5","address_5")
    )).map(item=>(item.userID,item))


    /** 2) 流式数据: 用户发的tweet数据*/
    /** 数据示例： * eventTime：事件时间、retweetCount：转推数、language：语言、userID：用户ID、favoriteCount：点赞数、id：事件ID * {"eventTime": "2018-11-05 10:04:00", "retweetCount": 1, "language": "chinese", "userID": "user_1", "favoriteCount": 1, "id": 4909846540155641457} */

    val kafkaDStream=KafkaUtils.createDirectStream[String,String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String,String](Set("testTopic3"),kafkaParams)
    ).map(item=>parseJson(item.value())).map(item=>{ 
   
      val userID = item.getString("userID")
      val eventTime = item.getString("eventTime")
      val language= item.getString("language")
      val favoriteCount = item.getInteger("favoriteCount")
      val retweetCount = item.getInteger("retweetCount")
      (userID,(userID,eventTime,language,favoriteCount,retweetCount))
    })


    /** 3) 流与静态数据做Join (RDD Join 方式)*/
    kafkaDStream.foreachRDD(_.join(userInfo).foreach(println))

    ssc.start()
    ssc.awaitTermination()

  }

  /**json解析*/
  def parseJson(log:String):JSONObject={ 
   
    var ret:JSONObject=null
    try{ 
   
      ret=JSON.parseObject(log)
    }catch { 
   
      //异常json数据处理
      case e:JSONException => println(log)
    }
    ret
  }

}

stream_static_rdd_join.png

Broadcast Join 方式

思路：RDD遍历每一条数据，去匹配广播变量中的值。

package com.bigData.spark

import com.alibaba.fastjson.{ 
   JSON, JSONException, JSONObject}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{ 
   Level, Logger}
import org.apache.spark.{ 
   SparkConf, SparkContext}
import org.apache.spark.streaming.kafka010.{ 
   ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{ 
   Durations, StreamingContext}

/** * Author: Wang Pei * License: Copyright(c) Pei.Wang * Summary: * * Stream-Static Join * * spark 2.2.2 * */
case class UserInfo(userID:String,userName:String,userAddress:String)
object StreamStaticJoin2 { 
   
  def main(args: Array[String]): Unit = { 
   

    //设置日志等级
    Logger.getLogger("org").setLevel(Level.WARN)

    //Kafka 参数
    val kafkaParams= Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (true: java.lang.Boolean),
      "group.id" -> "testTopic3_consumer_v1")

    //spark环境
    val sparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.replace("$","")).setMaster("local[3]")
    val ssc = new StreamingContext(sparkConf,Durations.seconds(10))

    /** 1) 静态数据: 用户基础信息。 将用户基础信息广播出去。*/
    val broadcastUserInfo=ssc.sparkContext.broadcast(
      Map(
        "user_1"->UserInfo("user_1","name_1","address_1"),
        "user_2"->UserInfo("user_2","name_2","address_2"),
        "user_3"->UserInfo("user_3","name_3","address_3"),
        "user_4"->UserInfo("user_4","name_4","address_4"),
        "user_5"->UserInfo("user_5","name_5","address_5")
      ))


    /** 2) 流式数据: 用户发的tweet数据*/
    /** 数据示例： * eventTime：事件时间、retweetCount：转推数、language：语言、userID：用户ID、favoriteCount：点赞数、id：事件ID * {"eventTime": "2018-11-05 10:04:00", "retweetCount": 1, "language": "chinese", "userID": "user_1", "favoriteCount": 1, "id": 4909846540155641457} */
    val kafkaDStream=KafkaUtils.createDirectStream[String,String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String,String](List("testTopic3"),kafkaParams)
    ).map(item=>parseJson(item.value())).map(item=>{ 
   
      val userID = item.getString("userID")
      val eventTime = item.getString("eventTime")
      val language= item.getString("language")
      val favoriteCount = item.getInteger("favoriteCount")
      val retweetCount = item.getInteger("retweetCount")
      (userID,(userID,eventTime,language,favoriteCount,retweetCount))
    })


    /** 3) 流与静态数据做Join (Broadcast Join 方式)*/
    val result=kafkaDStream.mapPartitions(part=>{ 
   
      val userInfo = broadcastUserInfo.value
      part.map(item=>{ 
   
        (item._1,(item._2,userInfo.getOrElse(item._1,null)))})
    })

    result.foreachRDD(_.foreach(println))


    ssc.start()
    ssc.awaitTermination()

  }

  /**json解析*/
  def parseJson(log:String):JSONObject={ 
   
    var ret:JSONObject=null
    try{ 
   
      ret=JSON.parseObject(log)
    }catch { 
   
      //异常json数据处理
      case e:JSONException => println(log)
    }
    ret
  }

}

stream_static_rdd_join2.png

流与半静态数据Join

半静态数据指的是放在Redis等的数据，会被更新。

思路：RDD 每个Partition连接一次Redis,遍历Partition中每条数据，根据k,去Redis中查找v。

package com.bigData.spark

import com.alibaba.fastjson.{ 
   JSON, JSONException, JSONObject}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{ 
   Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{ 
   ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{ 
   Durations, StreamingContext}
import redis.clients.jedis.Jedis

/** * Author: Wang Pei * License: Copyright(c) Pei.Wang * Summary: * * Stream-Static Join * * spark 2.2.2 * */
object StreamStaicJoin3 { 
   
  def main(args: Array[String]): Unit = { 
   

    //设置日志等级
    Logger.getLogger("org").setLevel(Level.WARN)

    //Kafka 参数
    val kafkaParams= Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (true: java.lang.Boolean),
      "group.id" -> "testTopic3_consumer_v1")

    //spark环境
    val sparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.replace("$","")).setMaster("local[3]")
    val ssc = new StreamingContext(sparkConf,Durations.seconds(10))

    /** 1) 半静态数据: 用户基础信息,在Redis中*/
    /** HMSET user_1 userID "user_1" name "name_1" address "address_1" */
    /** HMSET user_2 userID "user_2" name "name_2" address "address_2" */


    /** 2) 流式数据: 用户发的tweet数据*/
    /** 数据示例： * eventTime：事件时间、retweetCount：转推数、language：语言、userID：用户ID、favoriteCount：点赞数、id：事件ID * {"eventTime": "2018-11-05 10:04:00", "retweetCount": 1, "language": "chinese", "userID": "user_1", "favoriteCount": 1, "id": 4909846540155641457} */

    val kafkaDStream=KafkaUtils.createDirectStream[String,String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String,String](Set("testTopic3"),kafkaParams)
    ).map(item=>parseJson(item.value())).map(item=>{ 
   
      val userID = item.getString("userID")
      val eventTime = item.getString("eventTime")
      val language= item.getString("language")
      val favoriteCount = item.getInteger("favoriteCount")
      val retweetCount = item.getInteger("retweetCount")
      (userID,(userID,eventTime,language,favoriteCount,retweetCount))
    })

    /** 3) 流与半静态数据做Join (RDD Join 方式)*/
    val result=kafkaDStream.mapPartitions(part=>{ 
   
      val redisCli=connToRedis("localhost",6379,3000,10)
      part.map(item=>{ 
   
        (item._1,(item._2,redisCli.hmget(item._1,"userID","name","address")))
      })
    })

    result.foreachRDD(_.foreach(println))


    ssc.start()
    ssc.awaitTermination()

  }

  /**json解析*/
  def parseJson(log:String):JSONObject={ 
   
    var ret:JSONObject=null
    try{ 
   
      ret=JSON.parseObject(log)
    }catch { 
   
      //异常json数据处理
      case e:JSONException => println(log)
    }
    ret
  }

  /**连接到redis*/
  def connToRedis(redisHost:String,redisPort:Int,timeout:Int,dbNum:Int): Jedis ={ 
   
    val redisCli=new Jedis(redisHost,redisPort,timeout)
    redisCli.connect()
    redisCli.select(dbNum)
    redisCli
  }

}

stream_static_join3.png

Stream-Stream Join

流与流Join。

思路：DStream Join DStream。

package com.bigData.spark

import com.alibaba.fastjson.{ 
   JSON, JSONException, JSONObject}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{ 
   Level, Logger}
import org.apache.spark.{ 
   SparkConf, SparkContext}
import org.apache.spark.streaming.kafka010.{ 
   ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{ 
   Durations, StreamingContext}

/** * Author: Wang Pei * License: Copyright(c) Pei.Wang * Summary: * * Stream-Stream Join * * spark 2.2.2 * */
object StreamStreamJoin { 
   
  def main(args: Array[String]): Unit = { 
   

    //设置日志等级
    Logger.getLogger("org").setLevel(Level.WARN)

    //Kafka 参数
    val kafkaParams1= Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (true: java.lang.Boolean),
      "group.id" -> "testTopic3_consumer_v1")

    val kafkaParams2= Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (true: java.lang.Boolean),
      "group.id" -> "testTopic4_consumer_v1")


    //spark环境
    val sparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.replace("$","")).setMaster("local[3]")
    val ssc = new StreamingContext(sparkConf,Durations.seconds(10))

    /** 1) 流式数据: 用户发的tweet数据*/
    /** 数据示例： * eventTime：事件时间、retweetCount：转推数、language：语言、userID：用户ID、favoriteCount：点赞数、id：事件ID * {"eventTime": "2018-11-05 10:04:00", "retweetCount": 1, "language": "chinese", "userID": "user_1", "favoriteCount": 1, "id": 4909846540155641457} */

    val kafkaDStream1=KafkaUtils.createDirectStream[String,String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String,String](List("testTopic3"),kafkaParams1)
    ).map(item=>parseJson(item.value())).map(item=>{ 
   
      val userID = item.getString("userID")
      val eventTime = item.getString("eventTime")
      val language= item.getString("language")
      val favoriteCount = item.getInteger("favoriteCount")
      val retweetCount = item.getInteger("retweetCount")
      (userID,(userID,eventTime,language,favoriteCount,retweetCount))
    })

    /** 2) 流式数据: 用户发的tweet数据*/
    /** 数据示例： * eventTime：事件时间、retweetCount：转推数、language：语言、userID：用户ID、favoriteCount：点赞数、id：事件ID * {"eventTime": "2018-11-05 10:04:00", "retweetCount": 1, "language": "chinese", "userID": "user_1", "favoriteCount": 1, "id": 4909846540155641457} */

    val kafkaDStream2=KafkaUtils.createDirectStream[String,String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String,String](List("testTopic4"),kafkaParams2)
    ).map(item=>parseJson(item.value())).map(item=>{ 
   
      val userID = item.getString("userID")
      val eventTime = item.getString("eventTime")
      val language= item.getString("language")
      val favoriteCount = item.getInteger("favoriteCount")
      val retweetCount = item.getInteger("retweetCount")
      (userID,(userID,eventTime,language,favoriteCount,retweetCount))
    })

    /** 3) Stream-Stream Join*/
    val joinedDStream = kafkaDStream1.leftOuterJoin(kafkaDStream2)

    joinedDStream.foreachRDD(_.foreach(println))

    ssc.start()
    ssc.awaitTermination()

  }

  /**json解析*/
  def parseJson(log:String):JSONObject={ 
   
    var ret:JSONObject=null
    try{ 
   
      ret=JSON.parseObject(log)
    }catch { 
   
      //异常json数据处理
      case e:JSONException => println(log)
    }
    ret
  }

}

stream_stream_join.png

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请联系我们举报，一经查实，本站将立刻删除。

发布者：全栈程序员-站长，转载请注明出处：https://javaforall.net/147404.html原文链接：https://javaforall.net

赞 (0)
全栈程序员-站长的头像全栈程序员-站长
0 0


相关推荐

  • rabbitmq集群搭建_集群搭建步骤

    rabbitmq集群搭建_集群搭建步骤

    rabbitmq集群搭建_集群搭建步骤MQ为什么要使用集群？单机版无法满足真实应用的要求，如果RabbitMQ服务器遇到内存崩溃、机器掉电或者主板故障等情况况，该怎么办？单台RabbitMQ服务器可以满足每秒1000条消息的吞吐量，那么如果应用需要RabbitMQ服务满足每秒10万条消息的吞吐量呢？购买昂贵的服务器来增强单机RabbitMQ务的性能显得不足，此时搭建一个RabbitMQ集群才是解决实际生产中问题的关键。准备环境虚拟机环境：VMware®Workstation16Pro虚拟机版本：16.1.2build-1

    全栈程序员-站长的头像 全栈程序员-站长
    2025年10月20日
    3
  • 彻底弄懂StringBuffer与StringBuilder的区别「建议收藏」

    彻底弄懂StringBuffer与StringBuilder的区别「建议收藏」

    彻底弄懂StringBuffer与StringBuilder的区别「建议收藏」一问道StringBuffer与StringBuilder的区别，张口就来StringBuffer是线程安全的，因为它相关方法都加了synchronized关键字，StringBuilder线程不安全。没错，确实如此，但是我们查看过源码会发现StringBuffer是从jdk1.0就开始了，StringBuilder是从jdk1.5开始的。于是我就产生这样一个疑问，既然已经有了StringBu…

    全栈程序员-站长的头像 全栈程序员-站长
    2022年6月28日
    25
  • 2018 PyCharm环境安装教程[通俗易懂]

    2018 PyCharm环境安装教程[通俗易懂]

    2018 PyCharm环境安装教程[通俗易懂]环境准备：1.到PyCharm官网下载PyCharm安装包。2.下载了window版本的双击安装包进行安装。3.自定义软件安装路径(建议路径中不要中文字符)如：F:\DevSoftware\PyCharm2018.14.创建桌面快捷方式并关联*.py文件。5.选择开始菜单文件夹(默认即可)，点击安装。6.耐心等待安装。7.安装完成，勾选立即运行PyCharm。8.选择是否导入开发环境配置文件，我…

    全栈程序员-站长的头像 全栈程序员-站长
    2022年8月27日
    2
  • 团队解散，我们该何去何从？

    团队解散，我们该何去何从？

    团队解散，我们该何去何从？写在最前：纯属吐槽，随笔，勿喷！就在前些天，下班回家的路上，看到群信息，说：听说、听说京东裁员了～，图片来源于网络也是在上上月，也一度被传的沸沸扬扬的：阿里、京东、华为相继被曝停止社招，新闻也是满天飞舞，不管是裁员，还是停止社招，这些事情没有落在亲身经历，没有落在自己身上我们都会觉得不痛不痒，毕竟一个旁观者，永远无法感受当事人的喜怒哀乐～。俗话说：人无远虑必有近忧，假如当你遇上裁员，…

    全栈程序员-站长的头像 全栈程序员-站长
    2022年5月19日
    52
  • 轻松搞懂【TF-IDF、word2vec、svm、cnn、textcnn、bilstm、cnn+bilstm、bilstm+attention实现】英文长文本分类[通俗易懂]

    轻松搞懂【TF-IDF、word2vec、svm、cnn、textcnn、bilstm、cnn+bilstm、bilstm+attention实现】英文长文本分类[通俗易懂]

    轻松搞懂【TF-IDF、word2vec、svm、cnn、textcnn、bilstm、cnn+bilstm、bilstm+attention实现】英文长文本分类[通俗易懂]项目来源：https://www.kaggle.com/c/word2vec-nlp-tutorial/之前我写过几篇博客：就这？word2vec+BiLSTM、TextCNN、CNN+BiLSTM、BiLSTM+Attention实现中英文情感分类代码详解就这？word2vec+SVM（支持向量机）实现中英文情感分类代码详解这两篇博客主要是基于中文进行情感分类的，那么本篇博客，我会以这个kaggle项目来介绍如何实现英文长文本情感分类。1实验数据本次数据集来源于kaggle项目“Bago

    全栈程序员-站长的头像 全栈程序员-站长
    2022年6月28日
    30
  • RGBA(0,0,0,0)调色

    RGBA(0,0,0,0)调色

    RGBA(0,0,0,0)调色

    全栈程序员-站长的头像 全栈程序员-站长
    2021年10月28日
    159

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

关注全栈程序员社区公众号