更多内容详见:https://github.com/pierre94/flink-notes 一、基础概念 主要是两种处理模式: Connect/Join Union 二、双流处理的方法 Connect...DataStream,DataStream → ConnectedStreams 连接两个保持他们类型的数据流，两个数据流被Connect之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化..., lowData => (lowData.id, "healthy") ) (ConnectedStreams → DataStream 功能与 map 一样，对 ConnectedStreams 中的每一个流分别进行...、flink启动的是客户端 import java.text.SimpleDateFormat import org.apache.flink.api.common.state....{ValueState, ValueStateDescriptor} import org.apache.flink.streaming.api.TimeCharacteristic import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction
如果想获取数据流中Watermark的时间戳，或者在时间上前后穿梭，需要使用ProcessFunction系列函数，它们是Flink体系中最底层的API，提供了对数据流更细粒度的操作权限。...Flink SQL是基于这些函数实现的，一些需要高度个性化的业务场景也需要使用这些函数。 ?...状态的介绍可以参考我的文章：Flink状态管理详解，这里我们重点讲解一下的使用ProcessFunction其他几个特色功能。...本文所有代码都上传到了我的github：https://github.com/luweizheng/flink-tutorials Timer的使用方法 我们可以把Timer理解成一个闹钟，使用前先在Timer...) 这是一个回调函数，当到了“闹钟”时间，Flink会调用onTimer，并执行一些业务逻辑。
architectural-architectural-design-architecture 在目前的工作中，我对Java中的Stream和Lambda表达式都使用得很多，之前也写了两篇文章来总结对应的知识...背景 在Java中，如果你尝试对null做函数调用，就会引发NullPointerException（NPE），NPE是Java程序开发中的最典型的异常，对于Java开发者来说，无论你是初出茅庐的新人和还工作多年的老司机...，NPE经常让他们翻车。...为了避免NPE，他们会加很多if判断语句，使得代码的可读性变得很差。 从软件设计的角度来看，null本身是没有意义的语义，这是一种对缺失变量值的错误的建模。...Optional的目的就在于此：通过类型系统让你的领域模型中隐藏的知识显式地体现在你的代码中。
其是从进来的流数据中选取的字段。 关键看这个虚拟key是什么类型。...KeyedStream, String> KeyedStream5 = map.keyBy(tuple -> tuple.f1); 我的博客即将同步至腾讯云
但是这样会有一个问题，当数据库中的 schema 发生变化时，也需要手动更新对应的 Flink 任务以保持类型匹配，任何不匹配都会造成运行时报错使作业失败。这个操作冗余且繁琐，体验极差。...实际上对于任何和 Flink 连接的外部系统都可能有类似的上述问题，在 1.11.0 中重点解决了和关系型数据库对接的这个问题。...示例 目前对于jdbc catalog，flink仅提供了postgres catalog，我们基于postgres的catalog讲解一下如何使用flink的catalog ， 引入pom <dependency...tEnv中，然后就可以用tEnv进行一些操作了。 ...以一个简单的方法listDatabases为例： 从元数据表pg_database中查询所有的tablename，然后去掉内置的数据库，也就是template0和template1，然后封装到一个list
flink-window 窗口 在流处理应用中，数据是连续不断的，因此我们不可能等到所有数据都到了才开始处理。...其作用是在 Trigger 触发窗口计算之后将窗口的 State 中的数据清除。...如果session之间的间隔大于指定的间隔，数据将会被划分到不同的session中。...(即使后面还有延迟的数据 解决的问题 Watermark的时间戳可以和Event中的EventTime 一致，也可以自己定义任何合理的逻辑使得Watermark的时间戳不等于Event中的EventTime...， Event中的EventTime自产生那一刻起就不可以改变了，不受Apache Flink框架控制， 而Watermark的产生是在Apache Flink的Source节点或实现的Watermark
先上张图整体了解Flink中的异步io ?...阿里贡献给flink的，优点就不说了嘛，官网上都有，就是写库不会柱塞性能更好 然后来看一下， Flink 中异步io主要分为两种 一种是有序Ordered 一种是无序UNordered 主要区别是往下游...Flink中被设计成operator中的一种，自然去OneInputStreamOperator的实现类中去找 于是来看一下AsyncWaitOperator.java ?...方法（也就是前面那个包装类中的CompleteableFuture）并且传入了一个结果 看下complete方法源码 ?...这里比较绕，先将接收的数据加入queue中，然后onComplete()中当上一个异步线程getFuture() 其实就是每个元素包装类里面的那个CompletableFuture,当他结束时（会在用户方法用户调用
Flink版本：1.11.2 编写单元测试是设计生产应用程序的基本任务之一。如果不进行测试，那么一个很小的代码变更都会导致生产任务的失败。...因此，无论是清理数据、模型训练的简单作业，还是复杂的多租户实时数据处理系统，我们都应该为所有类型的应用程序编写单元测试。下面我们将提供有关 Apache Flink 应用程序的单元测试指南。...Apache Flink 提供了一个强大的单元测试框架，以确保我们的应用程序在上线后符合我们的预期。 1....我们使用 Flink 提供的 TestHarness 类，这样我们就不必自己创建模拟对象。...out.collect(String.format("Timer triggered at timestamp %d", timestamp)); } } 我们需要测试 KeyedProcessFunction 中的两个方法
，也会影响整体 Checkpoint 的进度，在这一步我们需要能够查看某个 PID 对应 hotmethod，这里推荐两个方法： 1、 多次连续 jstack，查看一直处于 RUNNABLE 状态的线程有哪些...； 2、使用工具 AsyncProfile dump 一份火焰图，查看占用 CPU 最多的栈； 二、作业失败，如何使用检查点 只需要指定检查点路径重启任务即可 bin/flink run -s :checkpointMetaDataPath.../article/details/89641904 三、总结下flink作业异常中断的操作流程 1、找出作业对应的jobID 2、进入hdfs对应目录，找到目录下面最新的检查点目录 3、通过指定检查点目录的方式重新启动作业...待作业运行稳定，查看作业最初异常中断的原因，记录下来并总结思考如何解决和避免。 四、怎么屏蔽flink checkpoint 打印的info 日志？...在log4j或者logback的配置文件里单独指定org.apache.flink.runtime.checkpoint.CheckpointCoordinator的日志级别为WARN
Flink SQL 支持对动态表进行复杂灵活的连接操作。 有几种不同类型的连接来解决可能需要的各种语义查询。 默认情况下，连接顺序未优化。 表按照在 FROM 子句中指定的顺序连接。...由于时间属性是准单调递增的，因此 Flink 可以从其状态中移除旧值而不影响结果的正确性。 基于时间的JOIN 基于事件时间的JOIN 基于时间的JOIN允许对版本化表进行连接。...Flink 使用 SQL:2011 标准的 FOR SYSTEM_TIME AS OF 的 SQL 语法来执行这个操作。...这种连接的强大之处在于，当无法将表具体化为 Flink 中的动态表时，它允许 Flink 直接针对外部系统工作。 以下处理时时态表联接示例显示了应与表 LatestRates 联接的仅追加表订单。...Orders 表中包含来自 MySQL 数据库中的 Customers 表的数据。
戳更多文章： 1-Flink入门 2-本地环境搭建&构建第一个Flink应用 3-DataSet API 4-DataSteam API 5-集群部署 6-分布式缓存 7-重启策略 8-Flink中的窗口...9-Flink中的Time 1时间类型 Flink中的时间与现实世界中的时间是不一致的，在flink中被划分为**事件时间，摄入时间，处理时间**三种。...**Event Time** Event Time 是事件发生的时间，一般就是数据本身携带的时间。这个时间通常是在事件到达 Flink 之前就确定的，并且可以从每个事件中获取到事件时间戳。...因为 Ingestion Time 使用稳定的时间戳（在源处分配一次），所以对事件的不同窗口操作将引用相同的时间戳，而在 Processing Time 中，每个窗口操作符可以将事件分配给不同的窗口（基于机器系统时间和到达延迟...在 Flink 中，Ingestion Time 与 Event Time 非常相似，但 Ingestion Time 具有自动分配时间戳和自动生成水印功能。
1窗口类型 1. flink支持两种划分窗口的方式（time和count） 如果根据时间划分窗口，那么它就是一个time-window 如果根据数据划分窗口，那么它就是一个count-window...：countWindow(5) `count-sliding-window` 有重叠数据的数量窗口，设置方式举例：countWindow(5,3) 4. flink支持在stream上的通过key去区分多个窗口...在滑窗中，一个元素可以对应多个窗口。...Flink 的 DataStream API 提供了简洁的算子来满足常用的窗口操作，同时提供了通用的窗口机制来允许用户自己定义窗口分配逻辑。...所有代码，我放在了我的公众号，回复Flink可以下载 海量【java和大数据的面试题+视频资料】整理在公众号，关注后可以下载~ 更多大数据技术欢迎和作者一起探讨~
CDC简介 Canal CanalJson反序列化源码解析 CDC简介 CDC,Change Data Capture,变更数据获取的简称，使用CDC我们可以从数据库中获取已提交的更改并将这些更改发送到下游...这些变更可以包括INSERT,DELETE,UPDATE等， 用户可以在以下的场景下使用CDC： 使用flink sql进行数据同步,可以将数据从一个数据同步到其他的地方，比如mysql、elasticsearch...可以在源数据库上实时的物化一个聚合视图 因为只是增量同步，所以可以实时的低延迟的同步数据 使用EventTime join 一个temporal表以便可以获取准确的结果 flink 1.11 将这些changelog...testGroup', 'canal-json.ignore-parse-errors'='true' -- 忽略解析错误，缺省值false ); CanalJson反序列化源码解析 canal 格式也是作为一种flink...pageId=147427289 [2].https://flink.apache.org/news/2020/07/06/release-1.11.0.html#table-apisql-support-for-change-data-capture-cdc
前言 对于 Java 开发者来说，null 是一个令人头疼的类型，一不小心就会发生 NPE （空指针） 问题。也是 Java 语言为人诟病的一个重要原因之一。...在我们消除可恶的 NPE 问题之前我们要回顾一下 Java 中 null 的概念。 2....NPE 问题的解决 很多时候我们对数据是否存在有自己的期望，但是这种期望并不能直接被我们掌控，一个返回值为 null 所表达的意思并不明确过于模糊，往往通过是否判断为 null 来规避空指针问题。...Java 8 中的 Optional Java 8 中的 Optional 是一个可选值的包装类。它的意义不仅仅帮我们简化了 NPE 问题的处理，同时也是 Java 函数式编程的一个重要辅助。...因为入参是不可控的，你无法保证入参中的 Optional 是否为 null。这恰恰违背了 Optional 的本意。
前言 今天我们主要聊聊flink中的一个接口org.apache.flink.api.common.functions.AggregateFunction，这个类可以接在window流之后，做窗口内的统计计算...注意：除了这个接口AggregateFunction，flink中还有一个抽象类AggregateFunction：org.apache.flink.table.functions.AggregateFunction...，大家不要把这个弄混淆了，接口AggregateFunction我们可以理解为flink中的一个算子，和MapFunction、FlatMapFunction等是同级别的，而抽象类AggregateFunction...sql的功能为例讲解一下flink的aggregate算子，其实就是我们用程序来实现这个sql的功能。...所以这个函数的入参是IN类型，返回值是ACC类型 merge 因为flink是一个分布式计算框架，可能计算是分布在很多节点上同时进行的，比如上述的add操作，可能同一个用户在不同的节点上分别调用了add
常用方法 Flink Table 内置的聚合方法包括： sum()：求和 count()：计数 avg()：平均值 min()：最小值 max()：最大值 stddevPop()：计算整个波动总体的标准偏差...stddevSamp()：计算样本数据的标准偏差 varPop()：计算整个波动总体的方差 varSamp()：计算样本数据的方差 另外，Flink Table 还支持自定义聚合方法。...示例 示例： import org.apache.flink.table.api._ import org.apache.flink.table.api.bridge.scala._ import org.apache.flink.api.scala...MyCountAccumulator, id: Long) = acc.count += 1 } class MyCountAccumulator { var count: Long = 0L } } 该示例中展示了...Flink Table内置的count/sum/max/min/avg等聚合方法的使用，并在最后展示了如何使用自定义聚合函数。
那把锁 锁用于多线程安全场景下，在Flink中存在一把锁，被用于数据处理线程、定时器调用线程、checkpoint线程。...在StreamTask中定义了一个Object对象lock，通过使用synchronized方式进行同步，在task的初始化过程中该对象传给了SystemProcessingTimeService、StreamInputProcessor...定时器调用线程 Flink中有一个很重要的功能那就是定时器，窗口触发需要定时器、用户自定义注册定时器需要定时器，但是定时器又可以按照时间属性分为两种：事件时间语义下watermark推进触发的定时器、处理时间语义下定时调度的定时器...在processElement中可能会操作状态、在定时回调onTimer中也可能会操作状态，那么状态就是作为共享数据，为了保证数据的一致性，所以这里加了锁。...processElement存在状态数据的竞争，为了保证数据的一致性，在checkpoint过程中会存在锁竞争： //StreamTask中performCheckpoint方法 synchronized
一、 为什么需要Window 在流处理应用中，数据是连续不断的，有时我们需要做一些聚合类的处理，例如：在过去的1分钟内有多少用户点击了我们的网页。...，API中通过window （WindowsAssigner assigner）指定。...测试数据 信号灯编号和通过该信号灯的车的数量 9,3 9,2 9,7 4,9 2,6 1,5 2,3 5,7 5,4 需求1:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现5次进行统计...--基于数量的滚动窗口 需求2:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现3次进行统计--基于数量的滑动窗口 package com.flink.source import org.apache.flink.api.common.functions.MapFunction...// 需求2:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现3次进行统计 val result2 = socketMap.keyBy(_.sensorId).countWindow
关键词：数据抽象 内存管理 Flink的数据抽象 MemorySegment Flink作为一个高效的流框架，为了避免JVM的固有缺陷（java对象存储密度低，FGC影响吞吐和响应等），必然走上自主管理内存的道路...但是在JVM的世界中，如果一个方法是一个虚方法，那么每次调用时，JVM都要花时间去确定调用的到底是哪个子类实现的该虚方法（方法重写机制，不明白的去看JVM的invokeVirtual指令），也就意味着每次都要去翻方法表...ByteBuffer与NetworkBufferPool 在MemorySegment这个抽象之上，Flink在数据从operator内的数据对象在向TaskManager上转移，预备被发给下个节点的过程中...在这行代码中，Flink把对象调用该对象所属的序列化器序列化为字节数组。 数据流转过程 上一节讲了各层数据的抽象，这一节讲讲数据在各个task之间exchange的过程。 整体过程 看这张图： ?...Flink的背压机制也是借此实现。
scope provided：用于打包的时候不包含这个依赖，以便增强代码移植性。 A3 解决 ?
领取专属 10元无门槛券
手把手带您无忧上云
云服务器
即时通信 IM
ICP备案
对象存储
实时音视频
