2020年8月9日 (日) 10:37的版本

Flink实时统计目标:

1.编程读取kafka缓存数据

2.商品实时统计:

(1)每日销售额,每日访问流量实时统计

(2)每日每家门店总销售额实时统计

(3)每日商品销量实时统计,并选出10大热销商品

(实现过程中,时间单位设置为1~2分钟)

项目开发一

a)编程读取kafka缓存数据

b)每日访问流量,每日销售额实时统计

c)设计与创建mysql结果存储表

d)打包程序

e)提交任务

f)查看运行结果

3.1 创建mysql数据表

我们会将处理后的数据存入mysql数据表中,所以现在进入master节点下

(1)查看mysql是否是启动状态

#service mysqld status

mysqld (pid 1409) is running.

(2)输入以下代码进入mysql中

mysql -uroot -proot

create database fk_shop;  //创建数据存放库
use fk_shop; //进入数据库
create table visitcount_everyday(datetime varchar(20) ,visicount double);//创建点击流表
create table salevolume(datetime varchar(20),salevolume double);//创建销售额表
show tables; //查看表

3.2 根据mysql数据表结构在idea中创建mysqlsink

(1)在scala中创建一个class对象:里面包括实时统计的日期和点击量(放入objectclass包里)

更改StringAndDouble.class

package objectclass
class StringAndDouble(datetime:String,sale:Double) {
  def getDate()= {datetime}
  def getSale() ={sale}
}

(2)在scala中再创建一个class对象 (放入mysqlsink包里)

更改SaleSQLsink'

package mysqlsink
import java.sql._
import objectclass.StringAndDouble
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
class SaleSQLsink (table:String) extends RichSinkFunction[StringAndDouble] with 
Serializable {
 //定义mysql配置连接
 var conn:Connection=_
 //创建语句
 var ps:PreparedStatement=_
 //  定义连接的mysql配置参数
 var user="root"
 var password="root"
 var driver= "com.mysql.jdbc.Driver"  
 var url = "jdbc:mysql://192.168.128.130:3306/fk_shop"  //连接到master节点下的fk_shop数据库
//重写invoke方法
 override def invoke(value: StringAndDouble): Unit ={
   Class.forName(driver)//加载并注册驱动
   //执行语句d
   conn = DriverManager.getConnection(url,user,password)
   val sql = "insert into" + table + "values(?,?)"
   ps = conn.prepareStatement(sql)
   ps.setString(1,value.getDate())
   ps.setDouble(2,value.getSale())
   ps.executeUpdate()
   if(ps!=null){ps.close()}
   if(conn != null) conn.close()
 }
}

3.3 实现Flink数据处理

(1)在scala里面创建一个object对象 (放到flinkclass包里面)

Flink实时处理编程过程

package flinkclass
//导入相关包
import java.util.Properties
import mysqlsink.SaleSQLsink
import objectclass.StringAndDouble
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.api.common.serialization.SimpleStringSchema
object SaleVolumn {
  def main(args: Array[String]): Unit = {
    //初始化env
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //配置kafka参数
    val pro = new Properties()
    pro.setProperty("bootstrap.servers","192.168.128.130:9092")
    pro.setProperty("group.id","test")
    //创建数据源
    val stream = env.addSource(new FlinkKafkaConsumer[String]("shop",new SimpleStringSchema(),pro))
    //数据处理过程
    val data = stream.map(x => x.split(",")).filter(x => x.nonEmpty && x.length == 11)
  .map { x => if (x(7).contains("buy")) (x(x.length - 1), (x(5).toDouble, 1.0)) else 
(x(x.length - 1), (0.0, 1.0)) }.keyBy(0).timeWindow(Time.minutes(1)).reduce((x, y) => 
(x._1, (x._2._1 + y._2._1, x._2._2 + y._2._2)))
    //保存每日销售额到mysql表
    data.map(x=>new StringAndDouble(x._1,x._2._1)).addSink(new SaleSQLsink("salevolume"))
    //保存每日点击流到mysql表
    data.map(x=>new StringAndDouble(x._1,x._2._2)).addSink(new SaleSQLsink("visitcount_everyday"))
    env.execute()
  }
}

(2)运行flink程序

①在所有节点查看kafka的状态,如果运行则kill掉

jps
kill -9 3507

②必须先启动zookeeper 再启动kafka,不然启动无效

zkServer.sh stop 关闭zookeeper
zkServer.sh start  启动zookeeper
zkServer.sh status 查看zookeeper的状态

③在所有节点启动zookeeper后在master,slave1,slave2上启动kafka

/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.propertiec 
>/tmp/klog &

④Jps查看

⑤在master启动flink集群

/usr/local/flink/start-cluster.sh

⑥查看主题是否创建没有则创建shop主题

kafka-topics.sh –zookeeper 192.168.128.130:2181–list  查看已创建主题
kafka-topics.sh –zookeeper 192.168.128.130:2181 –delete –topic shop 删除shop主题
kafka-topics.sh –zookeeper 192.168.128.130:2181 –create –topic shop –partitions 1 
–replication -factor 1  创建shop主题

⑦在IDEA中启动flume采集方案时,把之间采集的数据删除掉, 检查模拟采集是否启动

crontab -e

⑧启动之前写好的flume采集方案

fume-ng agent -c /usr/local/flume/conf/ -f /usr/local/flume/conf/spool-memory- 
kafka.conf -n client -Dflume.root.logger=INFO,console

⑨终于到了IDEA中本地运行salelumn.scala