Flink-intervalJoin源码和并行度问题

这篇具有很好参考价值的文章主要介绍了Flink-intervalJoin源码和并行度问题。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1.源码

底层用的是connect

把两个流的数据先保存到状态中

 先判断有没有迟到,迟到就放到侧输出流

再根据范围找数据

然后根据上界删除数据

Flink-intervalJoin源码和并行度问题,Flink,flink,大数据

 Flink-intervalJoin源码和并行度问题,Flink,flink,大数据

 文章来源地址https://www.toymoban.com/news/detail-574837.html

package com.atguigu.gmall.realtime.test;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author 
 * @date 2023/7/8
 * 该案例演示了Flink的intervalJoin
 */
public class Test03_IntervalJoin {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(2);

        //TODO 2.检查点相关的设置(略)
        //TODO 3.从指定的网络端口获取员工数据,指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<Emp> empDS = env
            .socketTextStream("hadoop102", 8888)
            .map(
                listStr -> {
                    String[] fieldArr = listStr.split(",");
                    return new Emp(Integer.valueOf(fieldArr[0]), fieldArr[1],
                        Integer.valueOf(fieldArr[2]), Long.valueOf(fieldArr[3]));
                }
            ).assignTimestampsAndWatermarks(
                // WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3))
                WatermarkStrategy
                    .<Emp>forMonotonousTimestamps()
                    .withTimestampAssigner(
                        new SerializableTimestampAssigner<Emp>() {
                            @Override
                            public long extractTimestamp(Emp emp, long recordTimestamp) {
                                return emp.getTs();
                            }
                        }
                    )
            );
        empDS.print("emp:");
        //TODO 4.从指定的网络端口获取部门数据,指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<Dept> deptDS = env
            .socketTextStream("hadoop102", 8889)
            .map(
                lineStr -> {
                    String[] fieldArr = lineStr.split(",");
                    return new Dept(Integer.valueOf(fieldArr[0]), fieldArr[1], Long.valueOf(fieldArr[2]));
                }
            ).assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<Dept>forMonotonousTimestamps()
                    .withTimestampAssigner(
                        new SerializableTimestampAssigner<Dept>() {
                            @Override
                            public long extractTimestamp(Dept dept, long recordTimestamp) {
                                return dept.getTs();
                            }
                        }
                    )
            );

        deptDS.print("dept:");

        //TODO 5.使用intervalJoin关联员工和部门
        empDS
            .keyBy(Emp::getDeptno)
            .intervalJoin(deptDS.keyBy(Dept::getDeptno))
            .between(Time.milliseconds(-5),Time.milliseconds(5))
            .process(
                new ProcessJoinFunction<Emp, Dept, Tuple2<Emp,Dept>>() {
                    @Override
                    public void processElement(Emp emp, Dept dept, Context ctx, Collector<Tuple2<Emp, Dept>> out) throws Exception {
                        out.collect(Tuple2.of(emp,dept));
                    }
                }
            ).print(">>>");
        env.execute();
    }
}

2.并行度问题

多个上游,一个下游,下游取最小的

一个上游,多个下游,广播

多个上游,多个下游,先广播,再取最小的

注意:水位线会减1ms

有可能出现上游出现了某个水位线,但是不触发,是因为别的并行度没有数据,水位线是long的最小值

尽量在source处设置水位线

只有kafka source是多个并行度,其他source都是一个并行度

Flink-intervalJoin源码和并行度问题,Flink,flink,大数据

 Flink-intervalJoin源码和并行度问题,Flink,flink,大数据

 

到了这里,关于Flink-intervalJoin源码和并行度问题的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 【Flink精讲】Flink性能调优:CPU核数与并行度

    提交任务命令: bin/flink run -t yarn-per-job -d -p 5 指定并行度 -Dyarn.application.queue=test 指定 yarn 队列 -Djobmanager.memory.process.size=2048mb JM2~4G 足够 -Dtaskmanager.memory.process.size=4096mb 单个 TM2~8G 足够 -Dtaskmanager.numberOfTaskSlots=2 与容器核数 1core: 1slot 或 2core: 1slot -c com.atguigu.flin

    2024年04月11日
    浏览(46)
  • flink算子的并行度设置方法

    #flink算子的并行度设置方法 并行度(Parallelism)是flink中一个非常重要的概念,它主要是指一个算子可以被分的子任务数,通常越高就意味着算子计算速度越快。 如上图所示,map()算子的并行度为2,window()算子的并行度也为2,也可以说整个数据流的并行度就是2。并行度的设置

    2024年03月17日
    浏览(51)
  • Flink学习笔记(七)并行度详解

    一个Flink程序由多个任务(Source、Transformation和Sink)组成。一个任务由多个并行实例(线程)来执行,一个任务的并行实例(线程)数目被称为该任务的并行度。 Flink是一个分布式流处理框架,它基于TaskManager和Slot来实现任务的执行。TaskManager是Flink中负责运行任务的工作进程

    2024年02月09日
    浏览(41)
  • 十八、Flink自定义多并行Source

    1、概述 1)作用 自定义多并行的Source,即Source的并行度可以是1到多个。 2)实现 1.继承RichParallelSourceFunction,重写run()方法。 2、代码实现

    2024年02月08日
    浏览(35)
  • Flink TaskSlot,算子链和并行度

    Flink TaskSlot,算子链和并行度 (108条消息) Flink TaskSlot与并行度_flink slot和并行度的关系_保护我方胖虎的博客-CSDN博客

    2024年02月15日
    浏览(45)
  • Flink-任务槽和并行度的关系

    任务槽 和并行度都跟程序的并行执行有关,但两者是完全不同的概念 。简单来说 任务槽 是静态的概念 ,是指TaskManager具有的并发执行能力,可以通过参数taskmanager.numberOfTaskSlots进行配置;而并行度是动态概念,也就是TaskManager运行程序时实际使用的并发能力,可以通过参数

    2024年02月11日
    浏览(40)
  • 【大数据面试题】008 谈一谈 Flink Slot 与 并行度

    一步一个脚印,一天一道面试题 该文章有较多引用文章 https://zhuanlan.zhihu.com/p/572170629?utm_id=0 并行度 Parallelism 概念作用 并行度是作用于 算子 的单位。Flink 的每个算子都可以单独设置并行度。一般来说,并行度越大,处理能力越大,处理的就越快。 Slot 概念作用 Slot 是 Flink

    2024年02月19日
    浏览(40)
  • 【Flink系列二】如何计算Job并行度及slots数量

    接上文的问题 并行的任务,需要占用多少slot ? 一个流处理程序,需要包含多少个任务 首先明确一下概念 slot:TM上分配资源的最小单元,它代表的是资源(比如1G内存,而非线程的概念,好多人把slot类比成线程,是不恰当的) 任务(task):线程调度的最小单元,和java中的

    2024年01月18日
    浏览(39)
  • Flink的API分层、架构与组件原理、并行度、任务执行计划

            Apache Flink的API分为四个层次,每个层次都提供不同的抽象和功能,以满足不同场景下的数据处理需求。下面是这四个层次的具体介绍: CEP API: Flink API 最底层的抽象为有状态实时流处理。其抽象实现是Process Function,并且Process Function被  框架集成到了DataStream API中

    2024年02月05日
    浏览(45)
  • flink源码分析 - flink命令启动分析

    flink版本: flink-1.12.1 源码位置:  flink-dist/src/main/flink-bin/bin/flink flink命令源码: 首先讲第一段: 工作中,很多人喜欢用符号链接(软连接)去将原始命令链接到一个新的文件。 例如:  将 /home/aaa链接到/opt/soft/flink-1.12.1/bin/flink,  实际使用的时候就可以用 aaa去代替flink命令。 例如

    2024年01月18日
    浏览(39)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包