“Flink实时数据处理”的版本间的差异
Xiaolongyang(讨论 | 贡献) (→Flink实时统计目标:) |
Xiaolongyang(讨论 | 贡献) |
||
第25行: | 第25行: | ||
f)查看运行结果 | f)查看运行结果 | ||
− | == | + | ==创建mysql数据表== |
我们会将处理后的数据存入mysql数据表中,所以现在进入master节点下 | 我们会将处理后的数据存入mysql数据表中,所以现在进入master节点下 | ||
第44行: | 第44行: | ||
show tables; //查看表 | show tables; //查看表 | ||
− | == | + | ==根据mysql数据表结构在idea中创建mysqlsink== |
(1)在scala中创建一个class对象:里面包括实时统计的日期和点击量(放入objectclass包里) | (1)在scala中创建一个class对象:里面包括实时统计的日期和点击量(放入objectclass包里) | ||
第98行: | 第98行: | ||
} | } | ||
− | == | + | ==实现Flink数据处理== |
(1)在scala里面创建一个object对象 (放到flinkclass包里面) | (1)在scala里面创建一个object对象 (放到flinkclass包里面) |
2020年8月9日 (日) 10:38的版本
Flink实时统计目标:
1.编程读取kafka缓存数据
2.商品实时统计:
(1)每日销售额,每日访问流量实时统计
(2)每日每家门店总销售额实时统计
(3)每日商品销量实时统计,并选出10大热销商品
(实现过程中,时间单位设置为1~2分钟)
项目开发一
a)编程读取kafka缓存数据
b)每日访问流量,每日销售额实时统计
c)设计与创建mysql结果存储表
d)打包程序
e)提交任务
f)查看运行结果
创建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; //查看表
根据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() } }
实现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