1. 全栈程序员必看首页

Spark Streaming Join「建议收藏」

全栈程序员-用户IM • • 未分类

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

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

发布者：全栈程序员-用户IM，转载请注明出处：https://javaforall.cn/147408.html原文链接：https://javaforall.cn

【正版授权，激活自己账号】： Jetbrains全家桶Ide使用，1年售后保障，每天仅需1毛

【官方授权 正版激活】： 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

赞 (0)
全栈程序员-用户IM全栈程序员-用户IM
0 0
blank

相关推荐

  • FDD与TDD的区别

    FDD与TDD的区别

    FDD与TDD的区别LTE通常分为FDDLTE和TDDLTEFDD，频分双工(FrequencyDivisionDuplexing)我和你通信，像广播一样，只能我说你听，是单工；像对讲机一样，同一时间只能一方说，另一方听，就是半双工；如果双方可以同时说和听，就是全双工。TDD：收发共用一个射频频点，上、下行链路使用不同的时隙来进行通信FDD：收发使用不同的射频频点来进行通信…

    全栈程序员-用户IM 全栈程序员-用户IM
    2022年6月11日
  • 基于深度学习的人脸性别识别系统（含UI界面，Python代码）「建议收藏」

    基于深度学习的人脸性别识别系统（含UI界面，Python代码）「建议收藏」

    基于深度学习的人脸性别识别系统（含UI界面，Python代码）「建议收藏」摘要：人脸性别识别是人脸识别领域的一个热门方向，本文详细介绍基于深度学习的人脸性别识别系统，在介绍算法原理的同时，给出Python的实现代码以及PyQt的UI界面。在界面中可以选择人脸图片、视频进行检测识别，也可通过电脑连接的摄像头设备进行实时识别人脸性别；可对图像中存在的多张人脸进行性别识别，可选择任意一张人脸框选显示结果，检测速度快、识别精度高。博文提供了完整的Python代码和使用教程，适合新入门的朋友参考，完整代码资源文件请转至文末的下载链接。

    全栈程序员-用户IM 全栈程序员-用户IM
    2022年5月23日
  • vue2.0集成百度UE编辑器，上传图片报错！！！

    vue2.0集成百度UE编辑器，上传图片报错！！！

    vue2.0集成百度UE编辑器，上传图片报错！！！

    全栈程序员-用户IM 全栈程序员-用户IM
    2021年10月11日
  • 流量精灵刷流量的实例教程

    流量精灵刷流量的实例教程

    流量精灵刷流量的实例教程流量精灵大概是去年的这个时候出来的一款提升网站访问量的软件，由于软件比较优秀，当时我就及时关注这款软件，现在时隔一年的时候，流量精灵在低调的发展下，版本也更新到了2.1.7。而以前只是写了一下流量精灵的测评文章，感觉已不能满足大家的需求，所以今天为广大的新手及新站长朋友写上详细的使用教程。流量精灵的功能介绍：主要就是刷流量的。这里的流量主要是指网站的流量，网站流量简单一点说就是访…

    全栈程序员-用户IM 全栈程序员-用户IM
    2022年9月29日
  • 二阶有源带通滤波器滤波原理[通俗易懂]

    二阶有源带通滤波器滤波原理[通俗易懂]

    二阶有源带通滤波器滤波原理[通俗易懂]转载增加：请注意有源滤波器因为运放的有效识别电压和响应频率的影响，适用于低频信号的滤波，对于高频信号最好使用无缘滤波。名词解释一阶低通滤波器，包含一组RC构成的滤波器，将谐波过滤一次；两阶低通滤波器，包含两组RC构成的滤波器，将谐波过滤两次。同理三阶滤波。。。。以下转载地址http://blog.sina.com.cn/s/blog_61144c49

    全栈程序员-用户IM 全栈程序员-用户IM
    2022年5月2日
  • 用UltraISO制作支持windows 7的U盘启动盘建议收藏

    用UltraISO制作支持windows 7的U盘启动盘建议收藏

    用UltraISO制作U盘启动盘，有人写过，我也看过，不过依照网上的那些文章，成功的并不多，经过几次试验，在不同的主板环境下成功概率高的方法应该如下：1.UltraISO建议9.3以上2.制作要点

    全栈程序员-用户IM 全栈程序员-用户IM
    2021年12月20日

发表回复

您的电子邮箱地址不会被公开。

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