大数据（9h）FlinkSQL双流JOIN

news2024/12/22 19:42:05

文章目录

  • 1、环境
  • 2、Temporal Joins
    • 2.1、基于处理时间（重点）
      • 2.1.1、设置状态保留时间
    • 2.2、基于事件时间
  • 3、Lookup Join（重点）
  • 4、Interval Joins（基于间隔JOIN）

重点是Lookup Join和Processing Time Temporal Join，其它随意

1、环境

WIN10+IDEA2021+JDK1.8+本地MySQL8

<properties>
    <maven.compiler.source>8</maven.compiler.source>
    <maven.compiler.target>8</maven.compiler.target>
    <flink.version>1.13.6</flink.version>
    <scala.binary.version>2.12</scala.binary.version>
    <slf4j.version>2.0.3</slf4j.version>
    <log4j.version>2.17.2</log4j.version>
    <fastjson.version>2.0.19</fastjson.version>
    <lombok.version>1.18.24</lombok.version>
</properties>
<!-- https://mvnrepository.com/ -->
<dependencies>
    <!-- Flink -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!-- FlinkSQL -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-csv</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-json</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!-- 日志 -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>${slf4j.version}</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>${slf4j.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-to-slf4j</artifactId>
        <version>${log4j.version}</version>
    </dependency>
    <!-- JSON解析 -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>${fastjson.version}</version>
    </dependency>
    <!-- 简化JavaBean书写 -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>${lombok.version}</version>
    </dependency>
</dependencies>

2、Temporal Joins

2.1、基于处理时间（重点）

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Hi {
    public static void main(String[] args) {
        //创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        //创建流式表执行环境
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env);
        //双流
        DataStreamSource<Tuple2<String, Integer>> d1 = env.fromElements(
                Tuple2.of("a", 2),
                Tuple2.of("b", 3));
        DataStreamSource<P> d2 = env.fromElements(
                new P("a", 4000L),
                new P("b", 5000L));
        //创建临时视图
        tbEnv.createTemporaryView("v1", d1);
        tbEnv.createTemporaryView("v2", d2);
        //双流JOIN
        tbEnv.sqlQuery("SELECT * FROM v1 LEFT JOIN v2 ON v1.f0=v2.pid").execute().print();
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class P {
        private String pid;
        private Long timestamp;
    }
}

结果

+----+-------+-------+-------------+-------------+
| op |    f0 |    f1 |         pid |   timestamp |
+----+-------+-------+-------------+-------------+
| +I |     a |     2 |      (NULL) |      (NULL) |
| -D |     a |     2 |      (NULL) |      (NULL) |
| +I |     a |     2 |           a |        4000 |
| +I |     b |     3 |      (NULL) |      (NULL) |
| -D |     b |     3 |      (NULL) |      (NULL) |
| +I |     b |     3 |           b |        5000 |
+----+-------+-------+-------------+-------------+

2.1.1、设置状态保留时间

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;
import java.util.Scanner;

public class Hi {
    public static void main(String[] args) {
        //创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        //创建流式表执行环境
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env);
        //设置状态保留时间
        tbEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5L));
        //双流
        DataStreamSource<Tuple2<String, Long>> d1 = env.addSource(new AutomatedSource());
        DataStreamSource<String> d2 = env.addSource(new ManualSource());
        //创建临时视图
        tbEnv.createTemporaryView("v1", d1);
        tbEnv.createTemporaryView("v2", d2);
        //双流JOIN
        tbEnv.sqlQuery("SELECT * FROM v1 INNER JOIN v2 ON v1.f0=v2.f0").execute().print();
    }

    /** 手动输入的数据源（请输入a或b进行测试） */
    public static class ManualSource implements SourceFunction<String> {
        public ManualSource() {}

        @Override
        public void run(SourceFunction.SourceContext<String> sc) {
            Scanner scanner = new Scanner(System.in);
            while (true) {
                String str = scanner.nextLine().trim();
                if (str.equals("STOP")) {break;}
                if (!str.equals("")) {sc.collect(str);}
            }
            scanner.close();
        }

        @Override
        public void cancel() {}
    }

    /** 自动输入的数据源 */
    public static class AutomatedSource implements SourceFunction<Tuple2<String, Long>> {
        public AutomatedSource() {}

        @Override
        public void run(SourceFunction.SourceContext<Tuple2<String, Long>> sc) throws InterruptedException {
            for (long i = 0L; i < 999L; i++) {
                Thread.sleep(1000L);
                sc.collect(Tuple2.of("a", i));
                sc.collect(Tuple2.of("b", i));
            }
        }

        @Override
        public void cancel() {}
    }
}

测试结果

2.2、基于事件时间

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Hello {
    public static void main(String[] args) {
        //创建流和表的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env);
        //创建数据流，设定水位线
        tbEnv.executeSql("CREATE TABLE v1 (" +
                "  x STRING PRIMARY KEY," +
                "  y BIGINT," +
                "  ts AS to_timestamp(from_unixtime(y,'yyyy-MM-dd HH:mm:ss'))," +
                "  watermark FOR ts AS ts - INTERVAL '2' SECOND" +
                ") WITH (" +
                "  'connector'='filesystem'," +
                "  'path'='src/main/resources/a.csv'," +
                "  'format'='csv'" +
                ")");
        tbEnv.executeSql("CREATE TABLE v2 (" +
                "  x STRING PRIMARY KEY," +
                "  y BIGINT," +
                "  ts AS to_timestamp(from_unixtime(y,'yyyy-MM-dd HH:mm:ss'))," +
                "  watermark FOR ts AS ts - INTERVAL '2' SECOND" +
                ") WITH (" +
                "  'connector'='filesystem'," +
                "  'path'='src/main/resources/b.csv'," +
                "  'format'='csv'" +
                ")");
        //执行查询
        tbEnv.sqlQuery("SELECT * " +
                "FROM v1 " +
                "LEFT JOIN v2 FOR SYSTEM_TIME AS OF v1.ts " +
                "ON v1.x = v2.x"
        ).execute().print();
    }
}

打印结果

+----+---+------------+-------------------------+--------+------------+-------------------------+
| op | x |          y |                      ts |     x0 |         y0 |                     ts0 |
+----+---+------------+-------------------------+--------+------------+-------------------------+
| +I | a | 1666540800 | 2022-10-24 00:00:00.000 | (NULL) |     (NULL) |                  (NULL) |
| +I | b | 1666540803 | 2022-10-24 00:00:03.000 |      b | 1666540802 | 2022-10-24 00:00:02.000 |
| +I | c | 1666540806 | 2022-10-24 00:00:06.000 |      c | 1666540803 | 2022-10-24 00:00:03.000 |
+----+---+------------+-------------------------+--------+------------+-------------------------+

3、Lookup Join（重点）

Lookup Join是基于Processing Time Temporal Join的

详细链接：https://yellow520.blog.csdn.net/article/details/128070761

4、Interval Joins（基于间隔JOIN）

SQL

SELECT * FROM v1
LEFT JOIN v2 ON v1.x=v2.x AND
v1.y BETWEEN (v2.y - INTERVAL '2' SECOND) AND (v2.y + INTERVAL '1' SECOND)

Java

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Scanner;

import static org.apache.flink.table.api.Expressions.$;

public class Hi {
    public static void main(String[] args) {
        //创建流和表的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env);
        //创建数据流
        DataStreamSource<String> d1 = env.addSource(new ManualSource());
        DataStreamSource<String> d2 = env.addSource(new AutomatedSource());
        //创建动态表，并声明一个额外的字段来作为处理时间字段
        Table t1 = tbEnv.fromDataStream(d1, $("x"), $("y").proctime());
        Table t2 = tbEnv.fromDataStream(d2, $("x"), $("y").proctime());
        //创建临时视图
        tbEnv.createTemporaryView("v1", t1);
        tbEnv.createTemporaryView("v2", t2);
        //执行查询
        tbEnv.sqlQuery("SELECT * FROM v1 LEFT JOIN v2 ON v1.x=v2.x AND " +
                "v1.y BETWEEN (v2.y - INTERVAL '2' SECOND) AND (v2.y + INTERVAL '1' SECOND)"
        ).execute().print();
    }

    /** 手动输入的数据源 */
    public static class ManualSource implements SourceFunction<String> {
        public ManualSource() {}

        @Override
        public void run(SourceFunction.SourceContext<String> sc) {
            Scanner scanner = new Scanner(System.in);
            while (true) {
                String str = scanner.nextLine().trim();
                if (str.equals("STOP")) {break;}
                if (!str.equals("")) {sc.collect(str);}
            }
            scanner.close();
        }

        @Override
        public void cancel() {}
    }

    /** 自动输入的数据源 */
    public static class AutomatedSource implements SourceFunction<String> {
        public AutomatedSource() {}

        @Override
        public void run(SourceFunction.SourceContext<String> sc) throws InterruptedException {
            for (int i = 0; i < 999; i++) {
                Thread.sleep(801);
                sc.collect("a");
                sc.collect("b");
            }
        }

        @Override
        public void cancel() {}
    }
}

测试结果

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/51488.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【Ubuntu】修改ubuntu和windows双系统启动顺序

【Ubuntu】修改ubuntu和windows双系统启动顺序

目录一、问题描述二、背景知识1. GRUB是什么2. GRUB配置文件3./etc/default/grub 主配置文件二、问题分析三、解决方案1. 修改grub主配置文件2. 更新grub配置文件一、问题描述 UbuntuWindows双系统默认使用GRUB作为引导管理器&#xff0c;而且通常默认启动Ubuntu。这样过于死板…
阅读更多...
用Python分析了30000+《独行月球》影评数据，看看观众们怎么说~

用Python分析了30000+《独行月球》影评数据，看看观众们怎么说~

文章目录&#x1f3f3;️‍&#x1f308; 1. 导入模块&#x1f3f3;️‍&#x1f308; 2. Pandas数据处理2.1 读取数据2.2 数据大小2.3 查看索引、数据类型和内存信息&#x1f3f3;️‍&#x1f308; 3. Pyecharts数据可视化3.1 《独行月球》评分分布-13.2 《独行月球》评分分布…
阅读更多...
TypeScript学习笔记

TypeScript学习笔记

TypeScript学习笔记 TypeScript 与 JavaScript 的区别 TypeScript 是 JavaScript 的超集&#xff0c;扩展了 JavaScript 的语法。TypeScript 可处理已有的 JavaScript 代码&#xff0c;并只对其中的 TypeScript 代码进行编译。TypeScript 文件的后缀名 .ts &#xff08;.ts&am…
阅读更多...
Milvus的索引方式

Milvus的索引方式

索引方式&#xff1a; FLAT&#xff1a;准确率高&#xff0c; 适合数据量小 暴力求解相似。 IVF-FLAT&#xff1a;量化操作&#xff0c; 准确率和速度的平衡 IVF:inverted file 先对空间的点进行聚类&#xff0c;查询时先比较聚类中心距离&#xff0c;再找到最近的N个点。 IV…
阅读更多...
redis5.0集群搭建(两台服务器)

redis5.0集群搭建(两台服务器)

文章目录1. 前言2. 配置两台机器内网互联3. redis安装4. redis集群5.0之前和5.0之后版本的区别4.1 redis5.0之前的版本创建集群4.2 redis5.0之后的版本创建集群4.3 redis5.0之前的版本需要依赖ruby环境5. redis5.0集群搭建5.1 创建redis-cluster目录5.2 将之前的redis.conf拷贝…
阅读更多...
windbg使用教程

windbg使用教程

下载 https://learn.microsoft.com/en-us/windows-hardware/drivers/debugger/debugger-download-tools 安装&#xff0c;打开exe 我选择了下载 打开X64 Debuggers And Tools-x64_en-us.msi 要安装对应系统位数 不然打不开 安装完成后没有反应。还以为我弄错了呢&#…
阅读更多...
【学习笔记】深度学习入门：基于Python的理论与实现-神经网络的学习

【学习笔记】深度学习入门：基于Python的理论与实现-神经网络的学习

CONTENTS四、神经网络的学习4.1 从数据中学习4.2 Loss function4.3 数值微分4.4 梯度4.5 学习算法的实现四、神经网络的学习 4.1 从数据中学习 神经网络的特征就是可以从数据中学习。所谓“从数据中学习”&#xff0c;是指可以由数据自动决定权重参数的值。利用特征量和机器学…
阅读更多...
HBase 开发：使用Java操作HBase 第1关：创建表

HBase 开发：使用Java操作HBase 第1关：创建表

为了完成本关任务&#xff0c;你需要掌握&#xff1a;1.如何使用Java连接HBase数据库&#xff0c;2.如何使用Java代码在HBase中创建表。 如何使用Java连接HBase数据库 Java连接HBase需要两个类&#xff1a; HBaseConfigurationConnectionFactoryHBaseConfiguration 要连接HBase…
阅读更多...
SPARKSQL3.0-Spark兼容多版本Hive源码分析

SPARKSQL3.0-Spark兼容多版本Hive源码分析

一、前言 阅读本节需要先掌握Catalog基础知识 Spark对Hive的所有操作都是通过获取Hive元数据[metastore]帮助spark构建表信息从而调用HDFS-API对原始数据的操作&#xff0c;可以说Spark兼容多版本Hive就是在兼容Hive的Metastore 二、源码分析 在catalog一节中我们知道spark…
阅读更多...
二.maven常用功能点

二.maven常用功能点

maven常用功能点一&#xff1a;分模块开发设计二&#xff1a;聚合三&#xff1a;继承1.依赖的继承2.可继承的内容3.聚合与继承的区分四&#xff1a;属性1.自定义属性2.内置属性3.Setting属性4.java系统属性5.环境变量属性五&#xff1a;java配置文件引用pom属性六&#xff1a;多…
阅读更多...
基于PHP+MySQL大学生心理健康管理系统的设计与实现

基于PHP+MySQL大学生心理健康管理系统的设计与实现

随着时代的发展,大学生的数量与日预增但是相对的也出现了很多心理问题,大学生自杀等心理问题引起的问题屡见不鲜如何能过更好的培养大学生正确的心理健康问题是现在很多大学多面临的一个重要的问题。本系统主要是对共享自行车的信息进行管理。该系统的基本功能包括用户登录,用户…
阅读更多...
云小课｜云小课带你玩转可视化分析ELB日志

云小课｜云小课带你玩转可视化分析ELB日志

阅识风云是华为云信息大咖&#xff0c;擅长将复杂信息多元化呈现&#xff0c;其出品的一张图(云图说)、深入浅出的博文(云小课)或短视频(云视厅)总有一款能让您快速上手华为云。更多精彩内容请单击此处。 云日志服务支持可视化查看日志&#xff0c;当ELB日志接入云日志服务后&a…
阅读更多...
[附源码]计算机毕业设计springboot青栞系统

[附源码]计算机毕业设计springboot青栞系统

项目运行 环境配置&#xff1a; Jdk1.8 Tomcat7.0 Mysql HBuilderX&#xff08;Webstorm也行&#xff09; Eclispe&#xff08;IntelliJ IDEA,Eclispe,MyEclispe,Sts都支持&#xff09;。 项目技术&#xff1a; SSM mybatis Maven Vue 等等组成&#xff0c;B/S模式 M…
阅读更多...
野火FPGA进阶（3）：SDRAM读写控制器的设计与验证

野火FPGA进阶（3）：SDRAM读写控制器的设计与验证

文章目录第50讲&#xff1a;SDRAM读写控制器的设计与验证理论部分设计与实现1. sdram_ctrlsdram_initsdram_a_refsdram_writesdram_readsdram_arbitsdram_ctrl2. sdram_topfifo_ctrlsdram_top3. uart_sdramuart_rxuart_txfifo_readuart_sdram第50讲&#xff1a;SDRAM读写控制器…
阅读更多...
[附源码]JAVA毕业设计工程车辆动力电池管理系统（系统+LW）

[附源码]JAVA毕业设计工程车辆动力电池管理系统（系统+LW）

[附源码]JAVA毕业设计工程车辆动力电池管理系统&#xff08;系统LW&#xff09; 目运行 环境项配置&#xff1a; Jdk1.8 Tomcat8.5 Mysql HBuilderX&#xff08;Webstorm也行&#xff09; Eclispe&#xff08;IntelliJ IDEA,Eclispe,MyEclispe,Sts都支持&#xff09;。 …
阅读更多...
怎么在Windows下使用Makefile文件

怎么在Windows下使用Makefile文件

前言&#xff1a;如果熟悉在Linux开发的话&#xff0c;肯定知道Makefile文件的用处&#xff0c;它给我们带来很多的便利。在Linux系统下并不会像Windows那么多开发工具&#xff0c;在Windows下&#xff0c;只要在开发工具上点击一个按钮&#xff0c;就能将工程的所有源码进行编…
阅读更多...
365天深度学习训练营-第P2周：彩色图片识别

365天深度学习训练营-第P2周：彩色图片识别

目录 一、前言 二、我的环境 三、代码实现 1、数据下载以及可视化 2、CNN模型 3、训练结果可视化 4、随机图像预测 四、模型优化 1、CNN模型 2、VGG-16模型 3、Alexnet模型 4、Resnet模型 一、前言 >- **&#x1f368; 本文为[&#x1f517;365天深度学习训练营]…
阅读更多...
2022年钒电池行业研究报告

2022年钒电池行业研究报告

第一章 行业概况 钒电池&#xff08;Vanadium Redox Battery&#xff0c;缩写为VRB&#xff09;&#xff0c;全称为全钒氧化还原液流电池&#xff0c;是一种活性物质呈循环流动液态的氧化还原电池。钒电池可以作为大容量储能电站的电池&#xff0c;其工作原理如下&#xff1a;…
阅读更多...
Unity 2021 请求 Android 12 读取本地文件权限

Unity 2021 请求 Android 12 读取本地文件权限

目标 工具&#xff1a; Unity 2021.2.14c1f1Android 12 系统手机 目标&#xff1a;实现Unity打出来的Apk包能请求读写android手机本地文件权限 原理 在Android系统中&#xff0c;操作手机中不安全的数据时&#xff0c;需要配置相应的权限&#xff0c;只有经过用户许可才能…
阅读更多...
[附源码]JAVA毕业设计个人信息管理系统（系统+LW）

[附源码]JAVA毕业设计个人信息管理系统（系统+LW）

[附源码]JAVA毕业设计个人信息管理系统&#xff08;系统LW&#xff09; 目运行 环境项配置&#xff1a; Jdk1.8 Tomcat8.5 Mysql HBuilderX&#xff08;Webstorm也行&#xff09; Eclispe&#xff08;IntelliJ IDEA,Eclispe,MyEclispe,Sts都支持&#xff09;。 项目技术…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023