基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(四)实时计算需求及技术方案

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
云数据库 RDS MySQL Serverless,0.5-2RCU 50GB
简介: 基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(四)实时计算需求及技术方案

16:实时计算需求及技术方案

  • 目标了解实时计算需求及技术方案
  • 路径
  • step1:实时计算需求
  • step2:技术方案
  • 实施
  • 实时计算需求
  • 实时统计消息总量
select count(*) from tbname;
  • 实时统计各个地区发送消息的总量
select sender_area,count(*) from tbname group by sender_area;
  • 实时统计各个地区接收消息的总量
select receiver_area,count(*) from tbname group by receiver_area;
  • 实时统计每个用户发送消息的总量
select sender_account,count(*) from tbname group by sender_account;
  • 实时统计每个用户接收消息的总量
select receiver_account,count(*) from tbname group by receiver_account;
  • |
  • 构建实时统计报表
  • 技术方案
  • 实时采集:Flume
  • 实时存储:Kafka
  • 实时计算:Flink
  • 实时结果:MySQL / Redis
  • 实时报表:FineBI / JavaWeb可视化

  • 小结
  • 了解实时计算需求及技术选型

17:Flink的基本介绍

  • 目标了解Flink的功能、特点及应用场景
  • 路径
  • step1:功能
  • step2:特点
  • step3:应用
  • 实施
Apache Flink is a framework and distributed processing engine for stateful computations over unbounded and bounded data streams. Flink has been designed to run in all common cluster environments, perform computations at in-memory speed and at any scale.
  • 功能:可以基于任何普通的集群平台,对有界的数据流或者无界的数据流实现高性能的有状态的分布式实时计算
  • Flink DataSet:对有界数据进行批处理操作
  • Flink DataStream:对无界数据进行实时处理操作
  • Flink Table:基于DSL实现结构化数据处理
  • Flink SQL:基于SQL实现结构化数据处理
  • Flink Gelly:Flink的图计算库
  • Flink ML:Flink的机器学习库
  • 特点
  • 支持高吞吐、低延迟、高性能的流处理
  • 支持带有事件时间的窗口(Window)操作
  • 支持有状态计算的Exactly-once语义
  • 支持高度灵活的窗口(Window)操作,支持基于time、count、session,以及data-driven的窗口操作
  • 支持具有Backpressure功能的持续流模型
  • 支持基于轻量级分布式快照(Snapshot)实现的容错
  • 一个运行时同时支持Batch on Streaming处理和Streaming处理
  • Flink在JVM内部实现了自己的内存管理
  • 支持迭代计算
  • 支持程序自动优化:避免特定情况下Shuffle、排序等昂贵操作,中间结果有必要进行缓存
  • 应用:所有实时及离线数据计算场景
  • 数据分析应用:实时数据仓库
  • Batch analytics(批处理分析)
  • Streaming analytics(流处理分析)
  • 事件驱动类应用
  • 欺诈检测(Fraud detection)
  • 异常检测(Anomaly detection)
  • 基于规则的告警(Rule-based alerting)
  • 业务流程监控(Business process monitoring)
  • Web应用程序(社交网络)
  • 数据管道ETL
  • Periodic ETL和Data Pipeline
  • 公司
  • 小结
  • 了解Flink的功能、特点及应用场景

18:代码模块构建

  • 目标实现开发环境的代码模块构建
  • 实施
  • 导入包中代码到IDEA中

  • flink包:应用类包,用于存放实际的应用类
  • MoMoFlinkCount:用于实现对每个需求的统计计算
  • MySQLSink:用于将计算的结果写入MySQL
  • pojo包:实体类包,用于存放所有实体类
  • MoMoCountBean:用于封装统计分析的结果
private Integer id ;        //结果id
private Long moMoTotalCount ;   //总消息数
private String moMoProvince ;       //省份
private String moMoUsername ;       //用户
private Long moMo_MsgCount ;        //消息数
//结果类型:1-总消息数 2-各省份发送消息数 3-各省份接受消息数 4-每个用户发送消息数 5-每个用户接受消息数
private String groupType ;   
  • utils包:工具类包,用于存放所有工具类
  • HttpClientUtils:用于实现将经纬度地址解析为省份的工具类
public static String findByLatAndLng(String lat , String lng)
  • 参数:经度,维度
  • 返回值:省份
  • 小结
  • 实现开发环境的代码模块构建

19:省份解析工具类测试

  • 目标:了解省份解析的实现
  • 路径
  • step1:基本设计
  • step2:注册百度开发者
  • step3:测试省份解析
  • 实施
  • 基本设计
  • 业务场景:根据IP或者经纬度解析得到用户的国家、省份、城市信息
  • 方案一:离线解析库【本地解析,快,不精准】
  • 方案二:在线解析库【远程解析,并发限制,精准】
  • 注册百度开发者
https://api.map.baidu.com/reverse_geocoding/v3/?ak=您的ak&output=json&coordtype=wgs84ll&location=31.225696563611,121.49884033194  //GET请求
  • 注册开放平台,获取AK码:参考《附录三》
  • 测试省份解析
  • 注意:将代码中的AK码更改为自己的
package bigdata.itcast.cn.momo.online.utils;
  import com.alibaba.fastjson.JSONObject;
  import org.apache.http.HttpEntity;
  import org.apache.http.client.methods.CloseableHttpResponse;
  import org.apache.http.client.methods.HttpGet;
  import org.apache.http.impl.client.CloseableHttpClient;
  import org.apache.http.impl.client.HttpClients;
  import org.apache.http.util.EntityUtils;
  import javax.swing.text.html.parser.Entity;
  import java.io.IOException;
  import java.util.Map;
  public class HttpClientUtils {
      //传入经纬度, 返回查询的地区
      public static String findByLatAndLng(String lat , String lng){
          try {
              CloseableHttpClient httpClient = HttpClients.createDefault();
              String url = "http://api.map.baidu.com/reverse_geocoding/v3/?ak=l8hKKRCuX2zrRa93jneDrPmc2UspGatO&output=json&coordtype=wgs84ll&location="+lat+","+lng;
              System.out.println(url);
              //请求解析
              HttpGet httpGet = new HttpGet(url);
              //得到结果
              CloseableHttpResponse response = httpClient.execute(httpGet);
              //获取数据
              HttpEntity httpEntity = response.getEntity();
              //转换成JSON
              String json = EntityUtils.toString(httpEntity);
              //从JSON中返回省份
              Map<String,Object> result = JSONObject.parseObject(json, Map.class);
              if(result.get("status").equals(0)){
                  Map<String,Object> resultMap =  (Map<String,Object>)result.get("result");
                  resultMap = (Map<String, Object>) resultMap.get("addressComponent");
                  String province = (String) resultMap.get("province");
                  return province;
              }
          } catch (IOException e) {
              e.printStackTrace();
          }
          return null;
      }
      public static void main(String[] args) {
          //测试
          String sf = findByLatAndLng("43.921297","124.655376");
          System.out.println(sf);
      }
  }
  • 小结
  • 了解省份解析的实现

20:Flink代码解读

  • 目标了解Flink代码的基本实现
  • 路径
  • step1:消费Kafka
  • step2:实时统计分析
  • step3:实时更新结果到MySQL
  • 实施
  • 消费Kafka
//构建Kafka配置
Properties props = new Properties();
props.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
props.setProperty("group.id", "momo2");
//构建消费者
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<String>("MOMO_MSG", new SimpleStringSchema(),props);
//Flink加载消费者
DataStreamSource<String> streamSource = env.addSource(kafkaConsumer);
  • 实时统计分析
//todo:3. 进行转换统计操作:
//3.1: 统计总消息量
countTotalMsg(streamSource);
//3.2: 基于经纬度, 统计各省份发送消息量
countProvinceSenderMsg(streamSource);
//3.3: 基于经纬度, 统计各省份接收消息量
countProvinceReceiverMsg(streamSource);
//3.4: 统计各个用户, 发送消息量
countUserNameSenderMsg(streamSource);
//3.5: 统计各个用户, 接收消息量
countUserNameReceiverMsg(streamSource);
//5. 执行flink操作
env.execute("momoFlinkCount");
  • 实时更新结果到MySQL
streamOperator.addSink(new MysqlSink("2"));
if (status.equals("2")){
    String sql = "select * from momo_count where momo_groupType = '2' and momo_province= '"+value.getMoMoProvince()+"' ";
    ResultSet resultSet = stat.executeQuery(sql);
    boolean flag = resultSet.next();
    if(flag) {
        sql = "update momo_count set momo_msgcount= '"+value.getMoMo_MsgCount()+ "' where momo_groupType = '2' and momo_province= '"+value.getMoMoProvince()+"' ";
    }else {
        sql = "insert into momo_count( momo_province,momo_msgcount,momo_groupType) values ('"+value.getMoMoProvince()+"',"+value.getMoMo_MsgCount()+",'2') ";
    }
    stat.executeUpdate(sql);
}
  • 小结
  • 了解Flink代码的基本实现

21:Flink实时计算测试

  • 目标实现Flink实时分析测试
  • 路径
  • step1:MySQL准备
  • step2:运行测试
  • 实施
  • MySQL准备
  • 找到SQL文件

  • 运行SQL文件创建结果数据库、表

  • 运行测试
  • 启动Flink程序:运行MoMoFlinkCount
  • 启动Flume程序
cd /export/server/flume-1.9.0-bin
bin/flume-ng agent -c conf/ -n a1 -f usercase/momo_mem_kafka.properties -Dflume.root.logger=INFO,console
  • 启动模拟数据
java -jar /export/data/momo_init/MoMo_DataGen.jar \
/export/data/momo_init/MoMo_Data.xlsx \
/export/data/momo_data/ \
100
  • 观察MySQL结果

  • 小结
  • 实现Flink实时分析测试


目录
相关文章
|
1月前
|
消息中间件 缓存 关系型数据库
Flink CDC产品常见问题之upsert-kafka增加参数报错如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
16天前
|
SQL 存储 API
阿里云实时计算Flink的产品化思考与实践【下】
本文整理自阿里云高级产品专家黄鹏程和阿里云技术专家陈婧敏在 FFA 2023 平台建设专场中的分享。
110137 4
阿里云实时计算Flink的产品化思考与实践【下】
|
2月前
|
消息中间件 Kafka Apache
Flink 提供了与 Kafka 集成的官方 Connector,使得 Flink 能够消费 Kafka 数据
【2月更文挑战第6天】Flink 提供了与 Kafka 集成的官方 Connector,使得 Flink 能够消费 Kafka 数据
69 2
|
30天前
|
消息中间件 关系型数据库 MySQL
Flink CDC产品常见问题之用upsert的方式写入kafka失败如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
2月前
|
SQL 消息中间件 分布式数据库
flink sql问题之连接HBase报错如何解决
Apache Flink是由Apache软件基金会开发的开源流处理框架,其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。
184 0
|
2月前
|
分布式计算 资源调度 Hadoop
Flink报错问题之Sql往kafka表写聚合数据报错如何解决
Apache Flink是由Apache软件基金会开发的开源流处理框架,其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。
|
2月前
|
SQL 关系型数据库 分布式数据库
Flink报错问题之用flush方法写入hbase报错如何解决
Apache Flink是由Apache软件基金会开发的开源流处理框架,其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。
|
2月前
|
SQL 消息中间件 关系型数据库
Flink CDC数据同步问题之向kafka同步数据报错如何解决
Flink CDC数据同步是指利用Flink CDC实现不同数据源之间的实时数据同步任务;本合集旨在提供Flink CDC数据同步的操作指南、性能优化建议和常见问题处理,助力用户高效实施数据同步。
|
3月前
|
存储 SQL 分布式数据库
OceanBase X Flink 基于原生分布式数据库构建实时计算解决方案
OceanBase X Flink 基于原生分布式数据库构建实时计算解决方案
|
3月前
|
消息中间件 SQL Java
阿里云Flink-自定义kafka sink partitioner实践及相关踩坑记录
阿里云Flink-自定义kafka sink partitioner实践及相关踩坑记录

热门文章

最新文章