WM_大屏实时计算深度剖析 flink写es kafka cannal配置 暂时没用

第1章 大屏实时计算深度剖析 学习目标 目标1：了解实时计算的应用场景 目标2：实时流计算的快速入门（ Flink 的入门使用） 目标3：Flink接入技术体系的剖析（ hdfs，jdbc，kafka, socket） 目标4：Flink 数据处理引擎的实战(采用双十一大屏， 热销数据统计， 区域分类统计， cep复杂事件的 处理) 目标5：实时流数据处理的设计方案（技术栈使用， Spring Boot /Spring Cloud集成） 目标6：Flink的核心技术点（多流JOIN, 流式的聚合去重，CEP 事件处理，UV布隆过滤器实现） 1. 实时计算应用场景 1.1 智能推荐 什么是智能推荐？ 定义： 根据用户行为习惯所提供的数据， 系统提供策略模型，自动推荐符合用户行为的信息。 例举： 比如根据用户对商品的点击数据（时间周期，点击频次）， 推荐类似的商品； 根据用户的评价与满意度， 推荐合适的品牌； 根据用户的使用习惯与点击行为，推荐类似的资讯。 应用案例： 1. 小红书推荐系统 +微信study322 专业一手超清完整 全网课程 超低价格+微信study322 专业一手超清完整 全网课程 超低价格 2. 实时流处理3. Flink处理（新一代大数据处理引擎） 1.2 实时数仓 什么是实时数仓 数据仓库（Data Warehouse） ,可简写为DW或DWH，是一个庞大的数据存储集合，通过对各种业务数 据进行筛选与整合，生成企业的分析性报告和各类报表，为企业的决策提供支持。实时仓库是基于 Storm/Spark（Streaming）/Flink等实时处理框架，构建的具备实时性特征的数据仓库。 应用案例 分析物流数据， 提升物流处理效率。 +微信study322 专业一手超清完整 全网课程 超低价格阿里巴巴菜鸟网络实时数仓设计： 数仓分层处理架构（流式ETL）： ODS -> DWD -> DWS -> ADS ODS（Operation Data Store）：操作数据层， 一般为原始采集数据。 DWD(Data Warehouse Detail) ：明细数据层， 对数据经过清洗，也称为DWI。 DWS(Data Warehouse Service)：汇总数据层，基于DWD层数据， 整合汇总成分析某一个主题域的服 务数据，一般是宽表， 由多个属性关联在一起的表， 比如用户行为日志信息：点赞、评论、收藏等。 ADS(Application Data Store): 应用数据层， 将结果同步至RDS数据库中， 一般做报表呈现使用。 +微信study322 专业一手超清完整 全网课程 超低价格1.3 大数据分析应用 1. IoT数据分析 1) 什么是IoT 物联网是新一代信息技术，也是未来发展的趋势，英文全称为： Internet of things（IOT），顾名 思义， 物联网就是万物相联。物联网通过智能感知、识别技术与普适计算等通信感知技术，广泛 应用于网络的融合中，也因此被称为继计算机、互联网之后世界信息产业发展的第三次浪潮。 2) 应用案例 物联网设备运营分析： 华为Iot数据分析平台架构： +微信study322 专业一手超清完整 全网课程 超低价格2. 智慧城市 城市中汽车越来越多， 川流不息，高德地图等APP通过技术手段采集了越来越多的摄像头、车流 的数据。 但道路却越来越拥堵，越来越多的城市开始通过大数据技术， 对城市实行智能化管理。 2018年， 杭州采用AI智慧城市，平均通行速度提高15%，监控摄像头日报警次数高达500次，识 别准确率超过92%，AI智慧城市通报占全体95%以上，在中国城市交通堵塞排行榜， 杭州从中国 第5名降至57名。 3. 金融风控 风险是金融机构业务固有特性，与金融机构相伴而生。金融机构盈利的来源就是承担风险的风险溢 价。 金融机构中常见的六种风险：市场风险、信用风险、流动性风险、操作风险、声誉风险及法律风 险。其中最主要的是市场风险和信用风险。 线上信贷流程，通过后台大数据系统进行反欺诈和信用评估： +微信study322 专业一手超清完整 全网课程 超低价格4. 电商行业 用户在电商的购物网站数据通过实时大数据分析之后， 通过大屏汇总展示， 比如天猫的双11购物 活动，通过大屏， 将全国上亿买家的订单数据可视化，实时性的动态展示，包含总览数据，流式 TopN数据，多维区域统计数据等，极大的增强了对海量数据的可读性。 TopN排行： +微信study322 专业一手超清完整 全网课程 超低价格+微信study322 专业一手超清完整 全网课程 超低价格 区域统计：+微信study322 专业一手超清完整 全网课程 超低价格2. Flink快速入门 2.1 Flink概述 Flink是什么 Flink是一个面向数据流处理和批处理的分布式开源计算框架。 无界流VS有界流 任何类型的数据都可以形成流数据，比如用户交互记录， 传感器数据，事件日志等等。 Apache Flink 擅长处理无界和有界数据集。 精确的时间控制和有状态的计算，使得 Flink能够运行 任何处理无界流的应用。 流数据分为无界流和有界流。 1） 无界流：有定义流的开始，但没有定义流的结束， 会不停地产生数据，无界流采用的是流处 理方式。 2） 有界流：有定义流的开始， 也有定义流的结束， 需要在获取所有数据后再进行计算，有界流 采用的是批处理方式。 +微信study322 专业一手超清完整 全网课程 超低价格编程模型 DataSet 一般用来处理有界流数据。 DataStream一般用来处理无界流数据。 2.2 Flink基础案例 1. 环境搭建配置 POM配置 FLINK集成 2. 批处理案例 功能： 通过批处理方式，统计日志文件中的异常数量。 代码： public class BatchProcessorApplication { public static void main(String[] args) throws Exception { // 1. 定义运行环境 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); // 2. 读取数据源（日志文件） DataSource<String> logData = env.readTextFile("./data/order_info.log"); // 3. 清洗转换数据 logData.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>> () { @Override public void flatMap(String value, Collector<Tuple2<String, Integer>> collector) throws Exception { // 1) 根据正则， 提取每行日志的级别 Pattern pattern = Pattern.compile("\\[main\\](.*?)\\["); Matcher matcher = pattern.matcher(value); if(matcher.find()) { // 2) 如果匹配符合规则， 放置元组内 collector.collect(new Tuple2<String,Integer> (matcher.group(1).trim(), 1)); } +微信study322 专业一手超清完整 全网课程 超低价格3. 流处理案例 功能： 根据 IP统计访问次数 代码： 2.3 Flink部署配置 1. 安装配置JDK8环境 } }).groupBy(0).sum(1).print(); // 4. 根据日志级别， 汇总统计， 打印结果 } } publ ic class StreamProcessorApplication { public static void main(String[] args) throws Exception{ // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取Socket数据源 DataStreamSource<String> socketStr = env.socketTextStream("127.0.0.1", 9911, "\n"); // 3. 转换处理流数据 socketStr.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() { @Override public void flatMap(String value, Collector<Tuple2<String, Integer>> collector) throws Exception { // 根据分隔符解析数据 String[] arrValue = value.split("\t"); collector.collect(new Tuple2<String,Integer>(arrValue[0], 1)); } }).keyBy(0).timeWindow(Time.seconds(5)).sum(1).print().setParallelism(2); env.execute("accessLog"); } } [root@localhost ~]# java -version java version "1.8.0_181" Java(TM) SE Runtime Environment (build 1.8.0_181-b13) Java HotSpot(TM) 64-Bit Server VM (build 25.181-b13, mixed mode) +微信study322 专业一手超清完整 全网课程 超低价格2. 下载Flink安装包 官方地址 安装包 3. 安装配置 1） 解压 2）运行 主节点访问端口： vi conf/masters: 4. 访问控制台 http://10.10.20.132:8081/#/overview Available Task Slots： 有效任务槽数量 对应配置文件： vi conf/flflink-conf.yaml TaskManger与JobManager关系 tar -xvf flink-1.11.2-bin-scala_2.11.tgz bin/ start-cluster.sh localhost:8081 taskmanager.numberOfTaskSlots: 1 +微信study322 专业一手超清完整 全网课程 超低价格Client 用来提交任务给 JobManager，JobManager 分发任务给 TaskManager 去执行， TaskManager 会采用心跳的方式， 汇报任务的执行状态。 JobManager 负责整个 Flink 集群任务的调度以及资源的管理。 TaskManager 负责具体的任务执行和对应任务在每个节点上的资源申请和管理。 2.4 Flink任务提交 第一种方式： 界面提交 1. 修改代码配置 socket数据源连接，采用主机名称配置 2. 工程代码打包 POM文件增加打包插件 DataStreamSource<String> socketStr = env.socketTextStream("flink1", 9911, "\n"); <build> <plugins> <!-- 编译插件 --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.5.1</version> <configuration> <source>1.8</source> <target>1.8</target> <!--<encoding>${project.build.sourceEncoding} </encoding>--> </configuration> </plugin> <!-- 打jar包插件(会包含所有依赖) --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>2.3</version> <executions> <execution> +微信study322 专业一手超清完整 全网课程 超低价格注意，这里不能采用spring-boot-maven-plugin打包插件， 否则flflink不能正常识别。 3. 提交任务 上传Jar包 接下来，在flflink1节点上， 开启Socket交互端口9911 <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> < ! - - zi p -d learn_spark.jar META INF/* . R S A M E T A - I N F / * . D S A M E T A - I N F / * . S F - -> < exclude>META-INF/*.SF</exclude> < e x c l u d e > M E T A - I N F / * . D S A < / e x c l u d e > < e x c l u d e > M E T A - I N F / * . R S A < / e x c l u d e > </excludes> </filter> </filters> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTra nsformer"> <!-- 可以设置jar包的入口类(可选) --> <mainClass>com.itcast.flink.usage.stream.StreamProcessorApplication</mainCl ass> </transformer> </transformers> </configuration> </execution> </executions> </plugin> </plugins> </build> +微信study322 专业一手超清完整 全网课程 超低价格然后提交并执行任务 savepoint path： 容错机制中快照保存的路径。 4. 运行验证 nc发送一些数据， 在TaskManager当中可以查看输出结果。 第二种方式： 命令行提交 在flflink控制台清除原有的Job任务。 1. 上传Jar包 将Jar包上传至flflink服务器： 2. 提交任务 采用命令行方式提交任务： 3. 验证结果 发送一些数据并在控制台验证输出结果。 [root@flink1 flink-1.11.2]# nc -lk 9911 [root@flink1 examples]# ll total 81880 drwxr-xr-x. 2 root root 194 Sep 9 23:48 batch -rw-r--r--. 1 root root 83843774 Sep 26 05:57 flink-usage-1.0-SNAPSHOT.jar drwxr-xr-x. 2 root root 50 Sep 9 23:48 gelly drwxr-xr-x. 3 root root 19 Sep 9 23:48 python drwxr-xr-x. 2 root root 241 Sep 9 23:48 streaming drwxr-xr-x. 2 root root 209 Sep 9 23:48 table [root@flink1 flink-1.11.2]# bin/flink run -c com.itcast.flink.usage.stream.StreamProcessorApplication examples/flink usage-1.0-SNAPSHOT.jar Job has been submitted with JobID 4c127f68f6683e5a9342410d7b6540db +微信study322 专业一手超清完整 全网课程 超低价格3. Flink接入体系 3.1 Flink Connectors Flink 连接器包含数据源 输入 与汇聚 输出两部分。 Flink 自身内置了一些基础的连接器，数据源输入包含 文件、目录、Sock et以及 支持从 collections 和 iter ators 中读取数据；汇聚输出支持把数据写入文件、 标准输出（stdout ）、标准错误输出（stderr）和 socket。 官方地址 Flink还可以支持扩展的连接器，能够与第三方系统进行交互。目前支持以下系统 : Apache Kafka (source/sink) Apache Cassandra (sink) Amazon Kinesis Streams (source/sink) Elasticsearch (sink) Hadoop FileSystem (sink) RabbitMQ (source/sink) Apache NiFi (source/sink) Twitter Streaming API (source) Google PubSub (source/sink) JDBC (sink) 常用的是Kafka、ES、HDFS以及JDBC。 3.2 JDBC（读/写） Flink Connectors JDBC 如何使用？ 功能： 将集合数据写入数据库中 代码： public class JdbcConnectorApplication { public static void main(String[] args) throws Exception{ // 配置日志文件 System.setProperty("log4j.configurationFile", "log4j2.xml"); // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 创建集合数据 List arrs = new ArrayList<String>(); arrs.add("10.10.20.101\t1601297294548\tPOST\taddOrder"); arrs.add("10.10.20.102\t1601297296549\tGET\tgetOrder"); // 3. 读取集合数据, 写入数据库 env.fromCollection(arrs).addSink(JdbcSink.sink( // 配置SQL语句 +微信study322 专业一手超清完整 全网课程 超低价格数据表： 自定义写入数据源 功能：读取Socket数据， 采用流方式写入数据库中。 代码： "insert into t_access_log (ip, time, type, api) values (?,?,?,?)", (ps, value) -> { System.out.println("receive ==> " + value); // 解析数据 String[] arrValue = String.valueOf(value).split("\t"); for(int i=0; i<arrValue.length; i++) { // 新增数据 ps.setString(i+1, arrValue[i]); } } , / / JD BC 连 接 配 置 n e w Jd b c Co nn ec tionOptions.JdbcConnectionOptionsBuilder() .withUrl("jdbc:mysql://192.168.19.150:3306/flink? useS S L = f a l s e " ) .withDriverName("com.mysql.jdbc.Driver") .withUsername("root") .withPassword("654321") .build())); // 4. 执行任务 env.execute("job"); } } DROP TABLE IF EXISTS `t_access_log`; CREATE TABLE `t_access_log` ( `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键ID', `ip` varchar(32) NOT NULL COMMENT 'IP地址', `time` varchar(255) NULL DEFAULT NULL COMMENT '访问时间', `type` varchar(32) NOT NULL COMMENT '请求类型', `api` varchar(32) NOT NULL COMMENT 'API地址', PRIMARY KEY (`id`) ) ENGINE = InnoDB AUTO_INCREMENT=1; public class CustomSinkApplication { public static void main(String[] args) throws Exception{ // 配置日志文件 System.setProperty("log4j.configurationFile", "log4j2.xml"); // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取Socket数据源 DataStreamSource<String> socketStr = env.socketTextStream("localhost", 9911, "\n"); +微信study322 专业一手超清完整 全网课程 超低价格AccessLog: 测试数据： // 3. 转换处理流数据 SingleOutputStreamOperator<AccessLog> outputStream = socketStr.map(new MapFunction<String, AccessLog>() { @Override public AccessLog map(String value) throws Exception { System.out.println(value); // 根据分隔符解析数据 String[] arrValue = value.split("\t"); // 将数据组装为对象 A c ce s s Lo g l o g = new AccessLog(); lo g .s e t N u m ( 1 ) ; f o r (i nt i= 0 ; i <a rrValue.length; i++) { if ( i == 0) { l og .s et I p (a r rV alue[i]); } e l se if ( i= = 1) { log.setTime(arrValue[i]); }else if( i== 2) { log.setType(arrValue[i]); }else if( i== 3) { log.setApi(arrValue[i]); } } return log; } }); // 4. 配置自定义写入数据源 outputStream.addSink(new MySQLSinkFunction()); // 5. 执行任务 env.execute("job"); } } @Data public class AccessLog { private String ip; private String time; private String type; private String api; private Integer num; } 10.10.20.11 1603166893313 GET getOrder 10.10.20.12 1603166893314 POST addOrder +微信study322 专业一手超清完整 全网课程 超低价格自定义读取数据源 功能： 读取数据库中的数据， 并将结果打印出来。 代码： 3.3 HDFS（读/写） 通过Sink写入HDFS数据 功能： 将Socket接收到的数据， 写入至HDFS文件中。 代码： public class CustomSourceApplication { publi c static void main(String[] args) throws Exception { / / 1 . 创 建 运 行 环 境 S t re a m Ex ec ut i on Environment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 配置自定义MySQL读取数据源 DataStreamSource<AccessLog> dataStream = env.addSource(new MySQLSourceFunction()); // 3. 设置并行度 dataStream.print().setParallelism(1); // 4. 执行任务 env.execute("custom jdbc source."); } } public class HdfsSinkApplication { public static void main(String[] args) throws Exception{ // 配置日志文件 System.setProperty("log4j.configurationFile", "log4j2.xml"); // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取Socket数据源 DataStreamSource<String> socketStr = env.socketTextStream("localhost", 9911, "\n"); BucketingSink<String> sink = new BucketingSink<String> ("d:/tmp/hdfs"); sink.setBucketer(new DateTimeBucketer<>("yyyy-MM-dd--HHmm")); sink.setWriter(new StringWriter()) .setBatchSize(5*1024) // 设置每个文件的大小 +微信study322 专业一手超清完整 全网课程 超低价格POM依赖: 数据源模拟实现： .setBatchRolloverInterval(5*1000) // 设置滚动写入新文件的时间 .setInactiveBucketCheckInterval(30*1000) // 30秒检查一次不写入的文 件 .setInactiveBucketThreshold(60*1000); // 60秒不写入，就滚动写入新的 文件 socketStr.addSink(sink).setParallelism(1); // 5. 执行任务 env.execute("job"); } } <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-filesystem_2.11</artifactId> <version>1.11.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.8.1</version> </dependency> public class SocketSourceApplication { /** * 服务端的端口 */ private int port; /** * 初始化构造方法 * @param port */ public SocketSourceApplication(int port) { this.port = port; } /** * IP 访问列表 */ private static String[] accessIps = new String[]{"10.10.20.101", "10.10.20.102", "10.10.20.103"}; /** * 请求访问类型 */ private static String[] accessTypes = new String[] {"GET", "POST", "PUT"}; +微信study322 专业一手超清完整 全网课程 超低价格/** * 请求接口信息 */ private static String[] accessApis = new String[] {"addOrder", "getAccount", "getOrder"}; /** * Netty通讯服务启动方法 * @thr ows Exception */ pub l i c v o i d ru n S e r v e r() throws Exception { / / 1 . 创 建 N e t t y 服 务 / / 2 . 定 义 事 件 B o s s 监 听 组 Ev e nt Lo o pG ro u p b o s s G r o up = ne w NioEventLoopGroup(); // 3. 定 义 用 来 处 理 已 经 被 接 收 的 连 接 EventLoopGroup workerGourp = new NioEventLoopGroup(); try { // 4. 定义NIO的服务启动类 ServerBootstrap sbs = new ServerBootstrap(); // 5. 配置NIO服务启动的相关参数 sbs.group(bossGroup, workerGourp) .channel(NioServerSocketChannel.class) // tcp最大缓存链接个数,它是tcp的参数, tcp_max_syn_backlog(半连 接上限数量, CENTOS6.5默认是128) .option(ChannelOption.SO_BACKLOG, 128) //保持连接的正常状态 .childOption(ChannelOption.SO_KEEPALIVE, true) // 根据日志级别打印输出 .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { //管道注册handler ChannelPipeline pipeline = socketChannel.pipeline(); //编码通道处理 pipeline.addLast("decode", new StringDecoder()); //转码通道处理 pipeline.addLast("encode", new StringEncoder()); // 处理接收到的请求 pipeline.addLast(new NettyServerHandler()); } }); System.err.println("-------server 启动------"); // 6. 监听控制台的输入, 并将输入信息, 广播发送给客户端 new Thread(new Runnable() { @Override public void run() { try { while(true) { String accessLog = getAccessLog(); System.out.println("broadcast (" + NettyServerHandler.channelList.size() + ") ==> " + accessLog); if(NettyServerHandler.channelList.size() > 0 ){ +微信study322 专业一手超清完整 全网课程 超低价格NettyServerHandler: for(Channel channel : NettyServerHandler.channelList) { channel.writeAndFlush(accessLog); } } Thread.sleep(1000); } }catch(Exception e) { e.printStackTrace(); } } }).start(); / / 7 . 启 动 n e t t y 服 务 C h an n e lF ut u r e c f = sbs.bind(port).sync(); cf.channel().closeFuture().sync(); }catch (Exception e) { e.printStackTrace(); } } /** * 获取访问日志 * @return */ private String getAccessLog() { StringBuilder strBuilder = new StringBuilder(); strBuilder.append(accessIps[new Random().nextInt(accessIps.length )]).append("\t") .append(System.currentTimeMillis()).append("\t") .append(accessTypes[new Random().nextInt(accessTypes.length)]).append("\t") .append(accessApis[new Random().nextInt(accessApis.length)]).append("\t\n"); return strBuilder.toString(); } /** * netty服务端的启动 * @param args * @throws Exception */ public static void main(String[] args) throws Exception{ new SocketSourceApplication(9911).runServer(); } } public class NettyServerHandler extends ChannelInboundHandlerAdapter { // 客户端通道记录集合 public static List<Channel> channelList = new ArrayList<>(); +微信study322 专业一手超清完整 全网课程 超低价格POM依赖： @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("Server---连接已建立: " + ctx); super.channelActive(ctx); // 将成功建立的连接通道, 加入到集合当中 channelList.add(ctx.channel()); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { System.out.println("Server---收到的消息: " + msg); } @O verride public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { System.out.println("server--读取数据出现异常"); cause.printStackTrace(); ctx.close(); } @Override public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { super.channelUnregistered(ctx); // 移除无效的连接通道 channelList.remove(ctx.channel()); } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { super.channelInactive(ctx); // 移除无效的连接通道 channelList.remove(ctx.channel()); } } <dependencies> <!-- Netty 核心组件依赖 --> <dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>4.1.16.Final</version> </dependency> <!-- spring boot 依赖 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> <version>${spring.boot.version}</version> </dependency> <!-- Spring data jpa 组件依赖--> <dependency> +微信study322 专业一手超清完整 全网课程 超低价格读取HDFS文件数据 Hadoop环境安装 1. 配置免密码登录 生成秘钥： <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-jpa</artifactId> <version>${spring.boot.version}</version> </dependency> <!-- mysql-connector-java --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> < v e r si o n >${mysql.jdbc.version}</version> </d e p e nd e n cy > <!-- Redis 缓存依赖 --> < d e p e nd e nc y > < g ro u p I d>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> <version>2.1.1.RELEASE</version> </dependency> </dependencies> public class HdfsSourceApplication { public static void main(String[] args) throws Exception{ // 配置日志文件 System.setProperty("log4j.configurationFile", "log4j2.xml"); // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取HDFS数据源 DataStreamSource<String> socketStr = env.readTextFile("hdfs://10.10.20.132:9090/hadoop-env.sh"); // 3. 打印文件内容 socketStr.print().setParallelism(1); // 4. 执行任务 env.execute("job"); } } [root@flink1 hadoop-2.6.0-cdh5.15.2]# ssh-keygen -t rsa -P '' Generating public/private rsa key pair. +微信study322 专业一手超清完整 全网课程 超低价格将秘钥写入认证文件： 修改认证文件权限： 2. 配置环境变量 将Hadoop 安装包解压， 将Hadoop加入环境变量/etc/profifile： 执行生效： 3. 修改Hadoop配置文件 1） 修改hadoop-env.sh文件 修改JAVA_HOME： 2）修改core-site.xml文件 这里的主机名称是flflink1。 3）修改hdfs-site.xml文件 [root@flink1 .ssh]# cat id_rsa.pub >> ~/.ssh/authorized_keys [root@flink1 .ssh]# chmod 600 ~/.ssh/authorized_keys e x p o r t H A DO O P_ HO M E= / u s r/ lo c a l / h a do op-2.6.0-cdh5.15.2 e x p o r t P AT H =$ HA DO O P_ H O M E/ b i n : $ PA T H source /etc/profile vi /usr/local/hadoop-2.6.0-cdh5.15.2/etc/hadoop export JAVA_HOME=/usr/local/jdk1.8.0_181 <configuration> <property> <name>fs.defaultFS</name> <value>hdfs://flink1:9090</value> </property> </configuration> +微信study322 专业一手超清完整 全网课程 超低价格4 ）修改mapred-site.xml文件 5）修改slaves文件 这里配置的是单节点， 指向本机主机名称。 6）修改yarn-site.xml 4. 启动Hadoop服务 <configuration> <property> <name>dfs.replication</name> <value>1</value> </property> <property> <name>hadoop.tmp.dir</name> < v alue>/usr/local/hadoop-2.6.0-cdh5.15.2/tmp</value> < / p r o p e r t y> </configuration> <configuration> <property> <name>yarn.nodemanager.aux-services</name> <value>mapreduce_shuffle</value> </property> </configuration> flink1 <configuration> <!-- Site specific YARN configuration properties --> <property> <name>mapreduce.framework.name</name> <value>yarn</value> </property> </configuration> [root@flink1 sbin]# ./start-all.sh This script is Deprecated. Instead use start-dfs.sh and start-yarn.sh 20/09/27 19:22:37 WARN util.NativeCodeLoader: Unable to load native hadoop library for your platform... using builtin-java classes where applicable Starting namenodes on [flink1] flink1: setterm: $TERM is not defined. flink1: starting namenode, logging to /usr/local/hadoop-2.6.0- cdh5.15.2/logs/hadoop-root-namenode-flink1.out flink1: setterm: $TERM is not defined. +微信study322 专业一手超清完整 全网课程 超低价格上传一个文件， 用于测试： 5. 访问验证 flink1: starting datanode, logging to /usr/local/hadoop-2.6.0- cdh5.15.2/logs/hadoop-root-datanode-flink1.out Starting secondary namenodes [0.0.0.0] 0.0.0.0: setterm: $TERM is not defined. 0.0.0.0: starting secondarynamenode, logging to /usr/local/hadoop- 2.6.0-cdh5.15.2/logs/hadoop-root-secondarynamenode-flink1.out 20/09/27 19:22:53 WARN util.NativeCodeLoader: Unable to load native hadoop library for your platform... using builtin-java classes where applicable start i n g y a r n d a e m o n s start i n g r es o u r c e m a na ger, logging to /usr/local/hadoop-2.6.0- c d h 5 .1 5 .2 / lo g s / y a r n- r o ot - re so u r ce m a n ag er-flink1.out f li n k 1 : s et t e r m : $ T ER M is no t d e f i ne d . f li n k 1 : s ta r ti n g n o d e m a na g e r , lo g g in g t o / u s r/ local/hadoop-2.6.0- cd h 5 . 15 .2 / lo g s/ y a r n - r oo t - no d e ma n a ge r - f l i n k 1 .o ut hdfs dfs -put /usr/local/hadoop-2.6.0-cdh5.15.2/etc/hadoop/hadoop env.sh / +微信study322 专业一手超清完整 全网课程 超低价格3.4 ES（写） ES服务安装 1. 到官网下载地址下载6.8.1版本的gz压缩包， 不要下载最新版本， Spring Boot等项目可能未 及时更新支持。 2. 解压安装包 3. ElasticSear ch不能以Root身份运行， 需要单独创建一个用户 执行以上命令，创建一个名为elsearch用户， 并赋予目录权限。 4. 修改配置文件 vi confifig/elasticsearch.yml, 只需修改以下设置： 5. 指定JDK版本 最新版的ElasticSearch需要JDK11版本， 下载JDK11压缩包， 并进行解压。 修改环境配置文件 vi bin/elasticsearch-env 参照以下位置， 追加一行， 设置JAVA_HOME， 指定JDK11路径。 tar - xvf elasticsearch-6.8.1-linux-x86_64.tar.gz 1. g ro u p ad d e l s ea rc h 2 . us e ra d d e l se a rc h -g elsearch -p elasticsearch 3. chown -R elsearch:elsearch /usr/local/elasticsearch-6.8.1 #集群名称 cluster.name: my-application #节点名称 node.name: node-1 #数据存储路径 path.data: /usr/local/elasticsearch-6.8.1/data #日志存储路径 path.logs: /usr/local/elasticsearch-6.8.1/logs # 绑定IP地址 network.host: 10.10.20.28 # 指定服务访问端口 http.port: 9200 # 指定API端户端调用端口 transport.tcp.port: 9300 +微信study322 专业一手超清完整 全网课程 超低价格关闭ConcMarkSweepGC JDK9版本以后不建议使用ConcMarkSweepGC， 如果不想出现提示， 可以将其关闭 vi confifig/jvm.options 将UseConcMarkSweepGC注释： 6. 启动ElasticSearch 切换用户 su elsearch 以后台常驻方式启动 bin/elasticsearch -d 7. 问题处理 出现max virtual memory areas vm.max_map_count [65530] is too low, increase to at least 错误信息 修改系统配置： vi /etc/sysctl.conf 添加 vm.max_map_count=655360 执行生效 sysctl -p vi /etc/security/limits.conf 在文件末尾添加 * soft nofifile 65536 * hard nofifile 131072 * soft nproc 2048 JAVA_HOME=/usr/local/jdk_11 # now set the path to java if [ ! -z "$JAVA_HOME" ]; then JAVA="$JAVA_HOME/bin/java" else if [ "$(uname -s)" = "Darwin" ]; then # OSX has a different structure JA VA="$ES_HOME/jdk/Contents/Home/bin/java" e ls e JAVA="$ES_HOME/jdk/bin/java" fi f i ## GC configuration #-XX:+UseConcMarkSweepGC ... ## G1GC Configuration # NOTE: G1GC is only supported on JDK version 10 or later. # To use G1GC uncomment the lines below. #-XX:-UseConcMarkSweepGC ... +微信study322 专业一手超清完整 全网课程 超低价格* hard nproc 4096 elsearch soft nproc 125535 elsearch hard nproc 125535 重新切换用户即可： su - elsearch FLINK ES写入功能实现 功能： 将Sock et流数据， 写入至ES服务。 代码： public class ElasticSinkApplication { public static void main(String[] args) throws Exception{ // 配置日志文件 System.setProperty("log4j.configurationFile", "log4j2.xml"); // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取Socket数据源 DataStreamSource<String> socketStr = env.socketTextStream("localhost", 9911, "\n"); //3. 配置ES服务信息 List<HttpHost> httpHosts = new ArrayList<>(); httpHosts.add(new HttpHost("10.10.20.132", 9200, "http")); //4. 数据解析处理 ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>( httpHosts, new ElasticsearchSinkFunction<String>() { public IndexRequest createIndexRequest(String element) { Map<String, String> json = new HashMap<>(); // 解析数据 String[] arrValue = String.valueOf(element).split("\t"); for(int i=0; i<arrValue.length; i++) { if(i == 0) { json.put("ip", arrValue[i]); }else if( i== 1) { json.put("time", arrValue[i]); }else if( i== 2) { json.put("type", arrValue[i]); }else if( i== 3) { json.put("api", arrValue[i]); } } return Requests.indexRequest() .index("flink-es") .type("access-log") .source(json); } @Override +微信study322 专业一手超清完整 全网课程 超低价格查看index信息： http://10.10.20.132:9200/_cat/indices?v 查看具体数据： http://10.10.20.132:9200/flflink-es/_search?pretty 3.5 KAFKA（读/写） Kafka安装 1. 下载Kafka_2.12-1.1.1安装包 2. 将安装包解压 3. 修改kafka配置 只修改绑定IP， 因为是单节点， 其他按默认配置来。 如有多个IP地址， 绑定为对外访问的IP。 4. 启动zookeeper服务 kafka安装包内置了zookeeper，可以直接启动。 public void process(String element, RuntimeContext ctx, RequestIndexer indexer) { indexer.add(createIndexRequest(element)); } } ); // 5. ES的写入配置 esSinkBuilder.setBulkFlushMaxActions(1); e s S i n k B ui ld e r . se t Re st C l ie nt F ac tory( re st C l i e nt Bu i ld e r -> { restClientBuilder.setMaxRetryTimeoutMillis(5000); } ); // 6. 添加ES的写入器 socketStr.addSink(esSinkBuilder.build()); socketStr.print().setParallelism(1); // 7. 执行任务 env.execute("job"); } } tar -xvf kafka_2.12-1.1.1.tgz listeners=PLAINTEXT://10.10.20.132:9092 advertised.listeners=PLAINTEXT://10.10.20.132:9092 +微信study322 专业一手超清完整 全网课程 超低价格5. 启动kafka服务 Flink Kafka 读取功能 功能： 通过flflink读取kafka消息队列数据， 并打印显示。 代码： 通过kafka生产者命令测试验证： 扩展点：kafka消息的消费处理策略： bin/zookeeper-server-start.sh -daemon config/zookeeper.properties bin/kafka-server-start.sh -daemon config/server.properties publ ic class KafkaSourceApplication { public static void main(String[] args) throws Exception { // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 设置kafka服务连接信息 Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "10.10.20.132:9092"); properties.setProperty("group.id", "fink_group"); // 3. 创建Kafka消费端 FlinkKafkaConsumer kafkaProducer = new FlinkKafkaConsumer( "flink-source", // 目标 topic new SimpleStringSchema(), // 序列化 配置 properties); // kafkaProducer.setStartFromEarliest(); // 尽可能从最早的记录开始 // kafkaProducer.setStartFromLatest(); // 从最新的记录开始 // kafkaProducer.setStartFromTimestamp(...); // 从指定的时间开始（毫秒） // kafkaProducer.setStartFromGroupOffsets(); // 默认的方法 // 4. 读取Kafka数据源 DataStreamSource<String> socketStr = env.addSource(kafkaProducer); socketStr.print().setParallelism(1); // 5. 执行任务 env.execute("job"); } } bin/kafka-console-producer.sh --broker-list 10.10.20.132:9092 --topic flink source +微信study322 专业一手超清完整 全网课程 超低价格Flink Kafka 写入功能 功能： 将Sock et的流数据，通过flflink 写入kafka 消息队列。 代码： 通过kafka消费者命令测试验证： 控制消息的发送处理模式: // kafkaProducer.setStartFromEarliest(); // 尽可能从最早的记录开始 // kafkaProducer.setStartFromLatest(); // 从最新的记录开始 // kafkaProducer.setStartFromTimestamp(...); // 从指定的时间开始（毫秒） // kafkaProducer.setStartFromGroupOffsets(); // 默认的方法 public class KafkaSinkApplication { public static void main(String[] args) throws Exception { // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取Socket数据源 DataStreamSource<String> socketStr = env.socketTextStream("localhost", 9911, "\n"); // 3. Kakfa的生产者配置 FlinkKafkaProducer kafkaProducer = new FlinkKafkaProducer( "10.10.20.132:9092", // broker 列表 "flink-topic", // 目标 topic new SimpleStringSchema()); // 序列化 方式 // 4. 添加kafka的写入器 socketStr.addSink(kafkaProducer); socketStr.print().setParallelism(1); // 5. 执行任务 env.execute("job"); } } bin/kafka-console-consumer.sh --bootstrap-server 10.10.20.132:9092 --topic flink-topic +微信study322 专业一手超清完整 全网课程 超低价格提供了三种消息处理模式： S e m a n t i c . N O NE ： Flink 不会有任何语义的保证，产生的记录可能会丢失或重复。 S e m a n t i c .A T _ LEA S T_ON CE （默认设置）：类似 FlinkKafkaProducer010 版本中的 setFlushOnCheckpoint(true) ，这可以保证不会丢失任何记录（虽然记录可能会重复）。 Semantic.EXACTLY_ONCE ：使用 Kafka 事务提供精准一次的语义。无论何时，在使用事务 写入 Kafka 时，都要记得为所有消费 Kafka 消息的应用程序设置所需的 isolation.level （ read_committed 或 read_uncommitted - 后者是默认值）。 Kafka 的消息可以携带时间戳，指示事件发生的时间或消息写入 Kafka broker 的时间。 3.6 自定义序列化（Protobuf） 在实际应用场景中， 会存在各种复杂传输对象，同时要求较高的传输处理性能， 这就需要采用自定义 的序列化方式做相应实现， 这里以Protobuf为例做讲解。 功能： kafka对同一Topic的生产与消费，采用Protobuf做序列化与反序列化传输， 验证能否正常解析 数据。 1. 通过protobuf脚本生成JAVA文件 通过批处理脚本，生成JAVA文件： // 控制消息的操作模式 Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "10.10.20.132:9092"); FlinkKafkaProducer kafkaProducer = new FlinkKafkaProducer( "flink-topic", // 目标 topic new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()), properties, Fl in k Ka f k a P roducer.Semantic.EXACTLY_ONCE ) ; / / 序 列 化 s c h em a kafkaProducer.setWriteTimestampToKafka(true); syntax = "proto3"; option java_package = "com.itcast.flink.connectors.kafka.proto"; option java_outer_classname = "AccessLogProto"; // 消息结构定义 message AccessLog { string ip = 1; string time = 2; string type = 3; string api = 4; string num = 5; } +微信study322 专业一手超清完整 全网课程 超低价格注意， 路径要配置正确。 2. 自定义序列化实现 添加POM 依赖： AccessLog对象： CustomSerialSchema: @echo off for %%i in (proto/*.proto) do ( d:/TestCode/protoc.exe --proto_path=./proto --java_out=../java ./proto/%%i echo generate %%i to java file successfully! ) <dep endencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.11</artifactId> <version>1.11.2</version> </dependency> <dependency> <groupId>com.google.protobuf</groupId> <artifactId>protobuf-java</artifactId> <version>3.8.0</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-beans</artifactId> <version>5.1.8.RELEASE</version> </dependency> </dependencies> @Data public class AccessLog implements Serializable { private String ip; private String time; private String type; private String api; private Integer num; } /** * 自定义序列化实现（Protobuf） */ +微信study322 专业一手超清完整 全网课程 超低价格public class CustomSerialSchema implements DeserializationSchema<AccessLog>, SerializationSchema<AccessLog> { private static final long serialVersionUID = 1L; private transient Charset charset; public CustomSerialSchema() { this(StandardCharsets.UTF_8); } p u b l ic C u st o m S e r i a l Sc he m a (C ha r se t ch a r se t ) { t h i s .c h a r se t = c he c k No tN ul l ( c ha r se t ) ; } public Charset getCharset() { return charset; } /** * 反序列化实现 * @param message * @return */ @Override public AccessLog deserialize(byte[] message) { AccessLog accessLog = null; try { AccessLogProto.AccessLog accessLogProto = AccessLogProto.AccessLog.parseFrom(message); accessLog = new AccessLog(); BeanUtils.copyProperties(accessLogProto, accessLog); return accessLog; } catch (Exception e) { e.printStackTrace(); } return accessLog; } @Override public boolean isEndOfStream(AccessLog nextElement) { return false; } /** * 序列化处理 * @param element * @return */ @Override public byte[] serialize(AccessLog element) { AccessLogProto.AccessLog.Builder builder = AccessLogProto.AccessLog.newBuilder(); BeanUtils.copyProperties(element, builder); return builder.build().toByteArray(); } +微信study322 专业一手超清完整 全网课程 超低价格3. 通过flflink对kafka消息生产者的实现 /** * 定义消息类型 * @return */ @Override public TypeInformation<AccessLog> getProducedType() { return TypeInformation.of(AccessLog.class); } } public class KafkaSinkApplication { public static void main(String[] args) throws Exception { // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取Socket数据源 DataStreamSource<String> socketStr = env.socketTextStream("localhost", 9911, "\n"); // 3. 转换处理流数据 SingleOutputStreamOperator<AccessLog> outputStream = socketStr.map(new MapFunction<String, AccessLog>() { @Override public AccessLog map(String value) throws Exception { System.out.println(value); // 根据分隔符解析数据 String[] arrValue = value.split("\t"); // 将数据组装为对象 AccessLog log = new AccessLog(); log.setNum(1); for(int i=0; i<arrValue.length; i++) { if(i == 0) { log.setIp(arrValue[i]); }else if( i== 1) { log.setTime(arrValue[i]); }else if( i== 2) { log.setType(arrValue[i]); }else if( i== 3) { log.setApi(arrValue[i]); } } return log; } }); // 3. Kakfa的生产者配置 Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "10.10.20.132:9092"); FlinkKafkaProducer kafkaProducer = new FlinkKafkaProducer( "10.10.20.132:9092", // broker 列表 +微信study322 专业一手超清完整 全网课程 超低价格开启Kafka消费者命令行终端，验证生产者的可用性： 4. 通过flflink对kafka消息订阅者的实现 "flink-serial", // 目标 topic new CustomSerialSchema() // 序列化 方式 ); // 4. 添加kafka的写入器 outputStream.addSink(kafkaProducer); socketStr.print().setParallelism(1); / / 5 . 执 行 任 务 e n v .e x ec ut e ("job"); } } [root@flink1 kafka_2.12-1.1.1]# bin/kafka-console-consumer.sh --bootstrap server 10.10.20.132:9092 --topic flink-serial 1601649380422GET" getAccount 1601649381422POSTaddOrder 1601649382422POST" public class KafkaSourceApplication { public static void main(String[] args) throws Exception { // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 设置kafka服务连接信息 Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "10.10.20.132:9092"); properties.setProperty("group.id", "fink_group"); // 3. 创建Kafka消费端 FlinkKafkaConsumer kafkaProducer = new FlinkKafkaConsumer( "flink-serial", // 目标 topic new CustomSerialSchema(), // 自定义序列化 properties); // 4. 读取Kafka数据源 DataStreamSource<AccessLog> socketStr = env.addSource(kafkaProducer); socketStr.print().setParallelism(1); // 5. 执行任务 env.execute("job"); } +微信study322 专业一手超清完整 全网课程 超低价格通过flflink的kafka生产者消息的发送， 对消费者的功能做测试验证。 4. Flink大屏数据实战 4.1 双十一大屏数据 总览数据 总销售量/总销售金额 TopN: 热销商品/商品类目/商品PV/商品UV 区域/分类数据 不同区域销售排名 不同分类销售排名 4.2 Canal同步服务安装 1. 下载安装包 安装包 后台管理包 2. 解压 解压安装包： 解压管理包： } mkdir -p /usr/local/canal tar -xzvf canal.deployer-1.1.4.tar.gz -C /usr/local/canal/ +微信study322 专业一手超清完整 全网课程 超低价格3. 初始化管理数据库 导入初始化数据脚本： 4. 修改MySQL服务同步配置 编辑配置文件： 增加同步配置： 重启服务： 检查同步功能是否开启： 创建同步用户： 赋予同步所需权限： mkdir -p /usr/local/canal-admin tar -xvf canal.admin-1.1.4.tar.gz -C /usr/local/canal-admin mysql -uroot -p654321 < /usr/local/canal-admin/conf/canal_manager.sql vi / etc/my.cnf [mysqld] log-bin=mysql-bin # 开启 binlog binlog-format=ROW # 选择 ROW 模式 server_id=1 # MySQL ID服务标识 service mysqld restart mysql> show variables like 'log_bin'; +---------------+-------+ | Variable_name | Value | +---------------+-------+ | log_bin | ON | +---------------+-------+ 1 row in set (0.01 sec) mysql> FLUSH PRIVILEGES; mysql> CREATE USER canal IDENTIFIED BY 'canal'; mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; Query OK, 0 rows affected (0.00 sec) mysql> FLUSH PRIVILEGES; Query OK, 0 rows affected (0.00 sec) +微信study322 专业一手超清完整 全网课程 超低价格5. 修改后台管理配置文件 配置内容： 先启动后台管理服务， 再启动Canal服务， 后台管理服务启动命令： 访问：http://10.10.20.133:8089/ 登录： admin/123456 6. Canal服务配置 配置内容： 启动Canal服务: vi /usr/local/canal-admin/conf/application.yml server: port: 8089 spring: jackson : d a te - fo rm a t : yy y y-MM-dd HH:mm:ss t im e -z on e : G MT + 8 spri ng.datasource: address: 192.168.19.150:3306 database: canal_manager username: root password: 654321 driver-class-name: com.mysql.jdbc.Driver url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}? useUnicode=true&characterEncoding=UTF-8&useSSL=false hikari: maximum-pool-size: 30 minimum-idle: 1 /usr/local/canal-admin/bin/startup.sh vi /usr/local/canal/conf/canal_local.properties # register ip canal.register.ip = 10.10.20.133 # canal admin config canal.admin.manager = 10.10.20.133:8089 canal.admin.port = 11110 canal.admin.user = admin canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441 # admin auto register canal.admin.register.auto = true canal.admin.register.cluster = +微信study322 专业一手超清完整 全网课程 超低价格7. 后台管理配置 修改Server管理配置： 修改Instance 配置（如果没有， 则新建，载入模板即可）： regex同步配置规则： 常见例子： 1. 所有表：.* or .\.. 2. canal schema下所有表： canal\..* 3. canal下的以canal打头的表：canal\.canal.* 4. canal schema下的一张表：canal.test1 5. 多个规则组合使用：canal\..*,mysql.test1,mysql.test2 (逗号分隔) 4.3 热销商品统计 功能实现流程： 1. 订单数据源的实现 2. flflink代码功能实现 3. Flink 与 Spring Boot的集成 4. 测试验证，比对SQL: /usr/local/canal/bin/startup.sh local # 指向ZK服务地址 canal.zkServers = 10.10.20.132:2181 # Canal同 步 方 式 canal.ser ve rM ode = kafka # mq服 务 地 址 canal . mq .s ervers = 10.10.20.132:9092 # mysql 同步服务ID标识， 不要配置冲突 canal.instance.mysql.slaveId=112 # mysql 同步主节点连接配置 canal.instance.master.address=192.168.19.150:3306 # 数据库用户名 canal.instance.dbUsername=canal # 数据库用户密码 canal.instance.dbPassword=canal # 数据同步消息队列 canal.mq.topic=order_binlog # 修改需要同步的数据库 canal.instance.filter.regex=flink.t_order select goodsId, sum(execPrice * execVolume) as totalAmount from t_order where execTime < 时间窗口的结束时间戳 group by goodsId order by totalAmount desc +微信study322 专业一手超清完整 全网课程 超低价格5. 数据呈现 kibana服务安装 Kibana是一个针对 Elasticsearch的开源分析及可视化平台，用来搜索、查看交互存储在Elasticsearch索 引中的数据。 1. 到官网下载， Kibana 安装包 , 与之对应 6.8.1版本， 选择 Linux 64 位版本下载，并进行解压。 2. Kibana启动不能使用 r oot用户， 使用上面创建的 elsear ch 用户， 进行赋权： 3. 修改配置文件 vi confifig/kibana.yml , 修改以下配置： 4. 启动kibana 看到以下日志， 代表启动正常 如果出现启动失败的情况， 要检查集群各节点的日志， 确保服务正常运行状态。 4.4 区域分类统计 1. 增加订单地址信息数据源 2. 创建对应的表与实体 实体： OrderAddress BO： JoinOrderAddress（订单数据与地址数据的合并对象） BO： HotDimensionOrder（ES存储的映射对象）， 注意这里的ID唯一性， 如果是按省份统计， ID存储省份信息，如果是按地级市统计， ID则存储为市区信息。 3. 改造订单数据源， 增加缓存写入， 地址信息数据源增加缓存的读取。 chow n -R elsearch:elsearch kibana-6.8.1-linux-x86_64 1 # 服务端口 server.port: 5601 # 服务地址 server.host: "0.0.0.0" # elasticsearch服务地址， 填写集群所有节点地址， 之间用逗号分割 elasticsearch.hosts: ["http://10.10.20.28:9200", "http://10.10.20.29:9200", "http://10.10.20.30:9200"] ./kibana -q log [01:40:00.143] [info][listening] Server running at http://0.0.0.0:5601 +微信study322 专业一手超清完整 全网课程 超低价格4. 修改Canal的后台配置， 增加地址数据源的监听队列。 5. 区域双流统计的核心代码实现： 1) 增加双流的kafka配置， 每个流监听不同的数据队列。 2）每个流要加上时间水印， 设定时间窗， 设定值比后面聚合的时间窗稍小一些。 3）根据订单ID做join匹配。 4） 根据区域做汇总统计（省份、城市）。 5） 将数据写入至 ES。 6. 测试验证 验证SQL: 4.5 订单状态监控统计（CEP） 1. 增加订单支付流水数据源 2. 创建对应的表与实体 实体： OrderPayment BO： JoinOrderAddress 3. 修改Canal的后台配置， 增加地址数据源的监听队列。 4. 核心代码实现： 1) 实现订单支付流水数据源的监听处理。 2）定义CEP处理规则，解析出支付成功的订单。 5. 测试验证 检查订单状态是未支付 -》 已支付的数据 检查超时的数据： 初始状态为0， 指定时间之内没有已支付的数据。 6. 拓展实现， 热门商品统计排行，只统计支付成功的数据。 sele ct province, goodsId, sum(execPrice * execVolume) totalAmount from t_order odr left join t_order_address adr on odr.id = adr.orderId where odr.execTime < 时间窗结束时间 group by province, goodsId order by province, totalAmount desc select * from t_order_payment pay where exists ( select 1 from t_order_payment tmp where tmp.orderId = pay.orderId and tmp.status = 0 ) and pay.status = 1 +微信study322 专业一手超清完整 全网课程 超低价格4.6 商品UV统计 功能： 统计商品在一段时间内的UV（Unique Visitor） 核心代码： public cla ss ScreenUniqueVisitorProcessor { /** * 执 行flink任务处理 * @throws Exception */ public void executeFlinkTask() throws Exception { // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime); env.setParallelism(1); // 2. 读取Socket数据源 // DataStreamSource<String> socketStr = env.socketTextStream("localhost", 9911, "\n"); DataStreamSource<String> socketStr = env.readTextFile("./data/goods_access.log"); // 3. 数据解析转换处理 socketStr.flatMap(new FlatMapFunction<String, GoodsAccessLog>() { @Override public void flatMap(String value, Collector<GoodsAccessLog> out) throws Exception { // 获取Json中的data数据 // 根据分隔符解析数据 String[] arrValue = value.split("\t"); System.out.println("receive msg => " + value); // 将数据组装为对象 GoodsAccessLog log = new GoodsAccessLog(); for(int i=0; i<arrValue.length; i++) { if(i == 0) { log.setIp(arrValue[i]); }else if( i== 1) { log.setAccessTime(Long.valueOf(arrValue[i])); }else if( i== 2) { log.setEventType(arrValue[i]); }else if( i== 3) { log.setGoodsId(arrValue[i]); } } out.collect(log); } }) .filter(new FilterFunction<GoodsAccessLog>() { +微信study322 专业一手超清完整 全网课程 超低价格4.7 布隆过滤器 功能： 统计商品在一段时间内的UV（采用布隆过滤器） 核心代码： @Override public boolean filter(GoodsAccessLog value) throws Exception { return value.getEventType().equals("view"); } }) .keyBy(GoodsAccessLog::getGoodsId) .timeWindow(Time.seconds(10)) .process( new ProcessWindowFunction<GoodsAccessLog, Map<String, String> , String, TimeWindow>(){ @ O v e r r i d e p u b li c v o id process(String key, Context context, Iterable<G o odsAccessLog> elements, Collector<Map<String, String>> out) throws Exception { S et < S t r i n g > i pS e t = n ew H a s h S et < > () ; M a p < S t r i n g , S t r i n g > g o o d s UV = n ew Li nkedHashMap<>(); elements.forEach( log -> { ipSet.add(log.getIp()); }); goodsUV.put(key , context.window().getEnd() + ":" + ipSet.size()); out.collect(goodsUV); } }) .print("uv result").setParallelism(1); // 5. 执行任务 env.execute("job"); } } /** * 执行flink任务处理 * @throws Exception */ public void executeFlinkTask() throws Exception { // 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); // 2. 读取Socket数据源 // DataStreamSource<String> socketStr = env.socketTextStream("localhost", 9911, "\n"); DataStreamSource<String> socketStr = env.readTextFile("./data/goods_access.log"); // 3. 数据解析转换处理 socketStr.flatMap(new FlatMapFunction<String, GoodsAccessLog>() { +微信study322 专业一手超清完整 全网课程 超低价格@Override public void flatMap(String value, Collector<GoodsAccessLog> out) throws Exception { // 获取Json中的data数据 // 根据分隔符解析数据 String[] arrValue = value.split("\t"); // 将数据组装为对象 GoodsAccessLog log = new GoodsAccessLog(); f o r (i nt i= 0 ; i <a rrValue.length; i++) { if ( i == 0) { l og .s et I p (a r rV alue[i]); } e l se if ( i= = 1) { l og .s et A c ce s sT ime(Long.valueOf(arrValue[i])); } e l se if ( i == 2) { log.setEventType(arrValue[i]); }else if( i== 3) { log.setGoodsId(arrValue[i]); } } out.collect(log); } }) .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<GoodsAccessLog>(Time.seconds(0)) { @Override public long extractTimestamp(GoodsAccessLog element) { return element.getAccessTime(); } }) .filter(new FilterFunction<GoodsAccessLog>() { @Override public boolean filter(GoodsAccessLog value) throws Exception { return value.getEventType().equals("view"); } }) .keyBy(GoodsAccessLog::getGoodsId) .timeWindow(Time.minutes(30)) .trigger(new CustomWindowTrigger()) .process(new CustomUVBloom()) .keyBy(0) .timeWindow(Time.seconds(3)) .max(1) .print("uv result => ").setParallelism(1); // 5. 执行任务 env.execute("job"); } +微信study322 专业一手超清完整 全网课程 超低价格