Spark_Streaming-低调大师

Spark_Streaming

2017-11-28 588

练习例子
1。

package com.haiyang

import java.nio.charset.Charset

import org.apache.flume.api.RpcClientFactory
import org.apache.flume.event.EventBuilder
//flume 数据发送 productor端口 主要用于发送产生的Event
object FlumeMsgSender {
  val client =RpcClientFactory.getDefaultInstance("master",33333)
//客户端通过RPC协议工厂实现默认的主节点 以及端口
  def sendEvent(msg:String)={
//发送event的Body
    val event =EventBuilder.withBody(msg,Charset.forName("UTF-8"))
//追加event
    client.append(event)
  }

  def main(args: Array[String]): Unit = {
//产生时间并发送
    (1 to 100).foreach(x=>{
      sendEvent(s"hello flume--$x")
    })
    client.close()
  }
}


package com.haiyang

import org.apache.spark.SparkConf
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}


object FlumePushStreaming {
//连接Spark 创建Streaming  Secods（5）表示五秒产生一个RDD
  val conf =new SparkConf().setMaster("local[*]").setAppName("flume get")
  val ssc=new StreamingContext(conf,Seconds(5))

  def main(args: Array[String]): Unit = {
//创建FlumeUntils 连接Stream  ：监听的IP 和端口
    val flumeDstream =FlumeUtils.createStream(ssc,"192.168.6.168",33333)
    flumeDstream.flatMap(x=>new String(x.event.getBody.array()).split("\\s"))
      .map(x=>(x,1))
      .reduceByKey(_+_)
      .print()
//将得到的数据展平 计算个数
    ssc.start()
    ssc.awaitTermination()
  }

}

//自建的工具类连接Hbase的配置文件连接HBase上的表明


package com.haiyang

import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.ConnectionFactory

object HbaseUtils {


  val conf =HBaseConfiguration.create()
  val connection =ConnectionFactory.createConnection(conf)

  def  getTable(tableName:String)={
    connection.getTable(TableName.valueOf(tableName))
  }

}

package com.haiyang

import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Duration, StreamingContext}


object KafkaDirectWordCount {
//连接Spark 连接Streaming
  val conf =new SparkConf().setMaster("local[*]").setAppName("use direct get wordCount")
  val ssc=new StreamingContext(conf,Duration(3000))

  def main(args: Array[String]): Unit = {
//实现kafka配置需要的信息 bootstrap.server key 以及value的Deserializer
    val kafkaParams =Map(("bootstrap.servers","master:9092,slave1:9092,slave2:9092"),("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
      ,("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"),("group.id","kafkatest"),("enable.auto.commit","false"))

//    val topic =Set("forstreaming")
//    val consumerStrategies =ConsumerStrategies Subscribe[String,String](topic, kafkaParams)
//    val kafkaDStream =KafkaUtils.createDirectStream(ssc,LocationStrategies.PreferConsistent,consumerStrategies)
//   kafkaDStream.map(x=>x.value())
//        .flatMap(x=>x.split("\\s"))
//              .map(_,1)
//          .reduceByKey(_+_)
//            .print()
//    ssc.start()
//    ssc.awaitTermination()
//发送topic 信息
    val topic =Set("test1")
    //Subscribe后面的两个泛型要与map的kv类型对应
 //consumerStrategies 消费者策略 就是连接到topic 并给定关于kafka参数信息 kafka 
   val 
consumerStrategies=ConsumerStrategies.Subscribe[String,String](topic,kafkaParams)
    val kafkaDstream=KafkaUtils.createDirectStream(ssc,LocationStrategies.PreferConsistent,consumerStrategies)
    kafkaDstream.map(x=>x.value())
      .flatMap(x=>x.split("\\s"))
      .map((_,1))
      .reduceByKey(_+_)
      .print()
    ssc.start()
    ssc.awaitTermination()

  }

}

package com.haiyang

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SpoolDirectoryWordCount {
//配置信息 连接SparkStreaming
  val conf =new SparkConf().setMaster("local[*]").setAppName("jianting")
  val ssc =new StreamingContext(conf,Seconds(5))


  def monitorDirectory()={
//监听hdfs上的文件夹
    val fileDstream =ssc.textFileStream("/bd17/stream_spark")
    fileDstream.print()
  }

  def main(args: Array[String]): Unit = {
    monitorDirectory()
    ssc.start()
    ssc.awaitTermination()
  }
}


package com.haiyang

import com.haiyang.FlumePushStreaming.ssc
import org.apache.spark.SparkConf
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

object StreamingPullFlume {
//连接SparkStreaming
  val conf =new SparkConf().setMaster("local[*]").setAppName(" pull data")
  val ssc= new StreamingContext(conf,Seconds(5))

  def main(args: Array[String]): Unit = {
 //从指定节点上拉去数据  
  val flumeDstream =FlumeUtils.createPollingStream(ssc,"master",9999)
    flumeDstream.map(x=>new String(x.event.getBody.array()))
      .flatMap(x=>x.split("\\s"))
      .map((_ ,1))
      .reduceByKey(_+_)
      .print()
//展平 做MR操作
    ssc.start()
    ssc.awaitTermination()

  }

}

package com.haiyang

import java.sql.DriverManager

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object StreamingSaveToMysql {

//创建连接SparkStreaming
  val conf =new SparkConf().setMaster("local[*]").setAppName("save to mysql")
  val ssc =new StreamingContext(conf,Seconds(5))

  def main(args: Array[String]): Unit = {
//通过socket连接指定主机的网络端口
    val dstream =ssc.socketTextStream("master",9999)
    val result =dstream.flatMap(x=>x.split("\\s"))
                    .map(x=>(x,1))
                      .reduceByKey(_+_)
//对ds数据践行遍历为RDD 键入元组 一个为声明时间戳 一个为遍历每个分区 让每个分区都连接到mysql上 在这里值得一提的是需要mysql依赖 
    result.foreachRDD((rdd,time)=>{
      val timestamp =time.milliseconds
      rdd.foreachPartition(wcs=>{
//启动驱动  连接mysql 
        Class.forName("com.mysql.jdbc.Driver")
        val connection =DriverManager.getConnection("jdbc:mysql://master:3306/test","ocean","BG@%pim%hGF1K5FY")
//sql语句 插入之前通过网络端口拉去到的数据
        val sql ="insert into streaming_wc (ts,word,count) values(?,?,?)"
        val prepareStatement =connection.prepareStatement(sql)
        for(record<-wcs){
          prepareStatement.setLong(1,timestamp)
          prepareStatement.setString(2,record._1)
          prepareStatement.setInt(3,record._2)
          prepareStatement.addBatch()
        }
        prepareStatement.executeBatch()
//        connection.commit()
        connection.close()
      })

    })
    ssc.start()
    ssc.awaitTermination()
  }

}

package com.haiyang


import org.apache.hadoop.hbase.client.Put
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable.ListBuffer

object StreamingToHbase {
//创建连接SparkStreaming
  val conf =new SparkConf().setAppName("streaming to hbase").setMaster("local[*]")
  val ssc =new StreamingContext(conf,Seconds(5))

  def main(args: Array[String]): Unit = {
//通过网端监听获得数据
val dstream =ssc.socketTextStream("master",9999)
    val result =dstream.flatMap(_.split("\\s"))
                          .map((_,1))
                            .reduceByKey(_+_)

//遍历为RDD 键入元组 两个参数 一个是时间戳 一个是word
    result.foreachRDD((rdd,time)=>{
      val timestamp =time.milliseconds.toString
      rdd.foreachPartition(wcs=>{
      val table =HbaseUtils.getTable("streaming_wc")

        val putList = new java.util.ArrayList[Put]()
        for(wc<-wcs){
          val put =new Put(timestamp.getBytes())
          put.addColumn("i".getBytes(),wc._1.getBytes(),wc._2.toString.getBytes())
          putList.add(put)
        }
        import  scala.collection.JavaConversions
        table.put(putList)
    })
    })
    ssc.start()
    ssc.awaitTermination()

  }


}

package com.haiyang

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Duration, StreamingContext}


object WindowWordCount {

  val conf =new SparkConf().setMaster("local[*]").setAppName("chuang kou caozuo")
  val ssc =new StreamingContext(conf,Duration(3000))

  def wordCount()= {
    val dstream = ssc.socketTextStream("master", 9999)
    val transaformation = dstream.flatMap(_.split("\\s"))
      .map((_, 1))

    //不指定滑动宽度  默认会以微批次的宽度计算的时间间隔
    val result = transaformation.reduceByKeyAndWindow((x1:Int,x2:Int)=>x1+x2, Duration(12000),Duration(6000))

    result.print()

  }
    def main(args: Array[String]): Unit = {
      wordCount()
      ssc.start()
      ssc.awaitTermination()
    }

}

package com.haiyang

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}


//从网络端口获得数据 并且统计每个单词出现的次数
object WordCount {
//如果使用local的话至少是local[2] 如果只有一个现成 那么这个现成始终再处理接收数据 计算数据的过程
val sparkconf =new SparkConf().setMaster("local[*]").setAppName("wc streaming")
val ssc=new StreamingContext(sparkconf,Seconds(3))


  def main(args: Array[String]): Unit = {
    val dsStream =ssc.socketTextStream("master",9999)

  val result =dsStream.flatMap(x=>x.split(" "))
            .map(x=>(x,1))
              .reduceByKey(_+_)
    result.print(20)
    ssc.start()
    ssc.awaitTermination()
  }




}

状态更新计算

package com.haiyang.statue

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Duration, StreamingContext}

object SteteWordCount {

  val conf =new SparkConf().setMaster("local[*]").setAppName("lei ji caozuo")
  val ssc=new StreamingContext(conf,Duration(3000))


  ssc.checkpoint("/temp/streamingcheckpoint")

  def allSumWordCount()={
    val dsream =ssc.socketTextStream("master",9999)

    val result =dsream.flatMap(_.split("\\s"))
                        .map((_,1))
                          .reduceByKey(_+_)
    //获取并更新状态
    val state =result.updateStateByKey[Int]((nowBat:Seq[Int],s:Option[Int])=>{
      s match{
        case Some(value)=>Some(value + nowBat.sum)
        case None=>Some(nowBat.sum)
      }
    })
    state.print()
  }

  def main(args: Array[String]): Unit = {
    allSumWordCount()
    ssc.start()
    ssc.awaitTermination()
  }
}

微信关注我们

原文链接：https://yq.aliyun.com/articles/277226

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

创建 kylin Module/Cube

1. 首先要创建 Project 2. 再把Hive 表加载进来： 3. 创建model 3.1. 首先选择或者创建一个project 3.2.创建一个新modle 3.3. 选择数据库 3.4 选择dimension (需要用到的列） 3.5. 选择 measure 用作过滤，聚合 6 4. 创建cube 4.1.选择model 4.2.输入cube名称自动生成dimension: 全选：这里没有用到聚合，所以直接到下一步： 4.3. build 在monitor 里查看进度: 最后在 Insight 里查询: 1 2 SELECT * FROM tb_dtu where DTUID = '4560477054' order by "TIME" DESC, ADDR 注意：TIME 等列要大写， TIME DESC 这里TIME要加引号本文转自疯吻IT博客园博客，原文链接：http://www.cnblogs.com/fengwenit/p/5826358.html，如需转载请自行联系原作者

2017-11-29

668

学hbase的时候，搭建好环境启动hbase，jps查看进程发现hmaster和hregionserevr可以正常出现。终端下输入hbase shell后也可以进入hbase的shell，输入List后就报错；错误提示：ERROR: org.apache.hadoop.hbase.PleaseHoldException: Master is initializing 查看错误日志发现：如图发现有一台节点无法连接，所以master一直是等待状态，不能列出具体的namespace.启动另一台集群的这个节点就恢复正常。当然要保证zookeeper能正常提供服务。可以通过查看zookeeper的状态来查看是否正常; zkServer.sh status; 必须保证半数以上的进程存活。然后就是要保证时间同步，可以临时先手动同步或者搭建NTP服务器。 hbase集群正常的截图本文转自 ChinaUnicom110 51CTO博客，原文链接:http://blog.51cto.com/xingyue2011/1934856

2017-11-29

672

资源下载

更多资源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。

Spark_Streaming

创建 kylin Module/Cube

HBase 报错 ERROR: org.apache.hadoop.hbase.PleaseHoldException: Master is...

相关文章

发表评论

资源下载

腾讯云软件源

Nacos

Rocky Linux

Sublime Text

欢迎您来访！