sparkstreaming测试之三有状态的接收数据-创新互联
测试思路:
创新互联是一家专业提供晋江企业网站建设,专注与网站设计制作、成都网站建设、H5建站、小程序制作等业务。10年已为晋江众多企业、政府机构等服务。创新互联专业网站设计公司优惠进行中。首先,使用上篇文章的程序一发送网络数据;
其次,运行spark程序,观察效果。
说明:
1. 这里使用到了更新函数;
2. 使用检查点来保证状态。
sparkStreaming import org.apache.log4j.{LoggerLevel} import org.apache.spark.streaming.{SecondsStreamingContext} import org.apache.spark.{SparkContextSparkConf} import org.apache.spark.streaming.StreamingContext._ object StatefulWordCount { def main(args:Array[]){ Logger.().setLevel(Level.) Logger.().setLevel(Level.) updateFunc = (values: []state:Option[]) => { currentCount = values.foldLeft()(_+_) previousCount = state.getOrElse() (currentCount + previousCount) } conf = SparkConf().setAppName().setMaster() sc = SparkContext(conf) ssc = StreamingContext(sc()) ssc.checkpoint() lines = ssc.socketTextStream(args()args().toInt) words = lines.flatMap(_.split()) wordCounts = words.map(x=>(x)) stateDstream = wordCounts.updateStateByKey[](updateFunc) stateDstream.print() ssc.start() ssc.awaitTermination() } }
另外有需要云服务器可以了解下创新互联scvps.cn,海内外云服务器15元起步,三天无理由+7*72小时售后在线,公司持有idc许可证,提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案,具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势,专为企业上云打造定制,能够满足用户丰富、多元化的应用场景需求。
当前名称:sparkstreaming测试之三有状态的接收数据-创新互联
分享URL:http://myzitong.com/article/ccigei.html