Flink 离线计算

文章目录

      • 一、样例一：读 csv 文件生成 csv 文件
      • 二、样例二：读 starrocks 写 starrocks
      • 三、样例三：DataSet、Table Sql 处理后写入 StarRocks
      • 四、遇到的坑

        <dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.9.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.11</artifactId><version>1.9.1</version></dependency><!--使用Java编程语言支持DataStream / DataSet API的Table＆SQL API--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java-bridge_2.11</artifactId><version>1.9.1</version><!--<scope>provided</scope>--></dependency><!--表程序规划器和运行时--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_2.11</artifactId><version>1.9.1</version><!--<scope>provided</scope>--></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-jdbc_2.11</artifactId><version>1.9.1</version><!--<scope>provided</scope>--></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.16.18</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.49</version></dependency>

一、样例一：读 csv 文件生成 csv 文件

  参考：（3）Flink学习- Table API & SQL编程

import lombok.Data;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.sinks.CsvTableSink;public class SQLWordCount {public static void main(String[] args) throws Exception {// 1、获取执行环境 ExecutionEnvironment （批处理用这个对象）ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();BatchTableEnvironment bTableEnv = BatchTableEnvironment.create(env);
//        DataSet<WC> input = env.fromElements(
//                WC.of("hello", 1),
//                WC.of("hqs", 1),
//                WC.of("world", 1),
//                WC.of("hello", 1)
//        );// 注册数据集
//        tEnv.registerDataSet("WordCount", input, "word, frequency");// 2、加载数据源到 DataSetDataSet<Student> csv = env.readCsvFile("D:\\tmp\\data.csv").ignoreFirstLine().pojoType(Student.class, "name", "age");// 3、将DataSet装换为TableTable students = bTableEnv.fromDataSet(csv);bTableEnv.registerTable("student", students);// 4、注册student表Table result = bTableEnv.sqlQuery("select name,age from student");result.printSchema();DataSet<Student> dset = bTableEnv.toDataSet(result, Student.class);System.out.println("count-->" + dset.count());dset.print();// 5、sink输出CsvTableSink sink1 = new CsvTableSink("D:\\tmp\\result.csv", ",", 1, FileSystem.WriteMode.OVERWRITE);String[] fieldNames = {"name", "age"};TypeInformation[] fieldTypes = {Types.STRING, Types.INT};bTableEnv.registerTableSink("CsvOutPutTable", fieldNames, fieldTypes, sink1);result.insertInto("CsvOutPutTable");env.execute("SQL-Batch");}@Datapublic static class Student {private String name;private int age;}
}

  准备测试文件 data.csv：

name,age
zhangsan,23
lisi,43
wangwu,12

  运行程序后会生成 D:\\tmp\\result.csv 文件。

二、样例二：读 starrocks 写 starrocks

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;public class SQLWordCount {public static void main(String[] args) throws Exception {TypeInformation[] fieldTypes = {Types.STRING, Types.INT};RowTypeInfo rowTypeInfo = new RowTypeInfo(fieldTypes);JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat().setDrivername("com.mysql.jdbc.Driver").setDBUrl("jdbc:mysql://192.168.xx.xx:9030/dwd?characterEncoding=utf8").setUsername("root").setPassword("").setQuery("select * from student").setRowTypeInfo(rowTypeInfo).finish();final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();// 方式一DataSource s = env.createInput(jdbcInputFormat);s.output(JDBCOutputFormat.buildJDBCOutputFormat().setDrivername("com.mysql.jdbc.Driver").setDBUrl("jdbc:mysql://192.168.xx.xx:9030/dwd?characterEncoding=utf8").setUsername("root").setPassword("").setQuery("insert into student values(?, ?)").finish());// 方式二
//        DataSet<Row> dataSource = env.createInput(jdbcInputFormat);
//
//        dataSource.output(JDBCOutputFormat.buildJDBCOutputFormat()
//                .setDrivername("com.mysql.jdbc.Driver")
//                .setDBUrl("jdbc:mysql://192.168.xx.xx:9030/dwd?characterEncoding=utf8")
//                .setUsername("root").setPassword("")
//                .setQuery("insert into student values(?, ?)")
//                .finish()
//        );env.execute("SQL-Batch");}
}

  数据准备：

CREATE TABLE student (name STRING,age INT
) ENGINE=OLAP 
DUPLICATE KEY(`name`)
DISTRIBUTED BY RANDOM
PROPERTIES (
"compression" = "LZ4",
"fast_schema_evolution" = "false",
"replicated_storage" = "true",
"replication_num" = "1"
);insert into student values('zhangsan', 23);

参考：
flink 读取mysql源 JDBCInputFormat、自定义数据源
flink1.10中三种数据处理方式的连接器说明
flink读写MySQL的两种方式

注意：如果运行 java -cp flink-app-1.0-SNAPSHOT-jar-with-dependencies.jar com.xiaoqiang.app.SQLWordCount 时报错：Exception in thread "main" com.typesafe.config.ConfigException$UnresolvedSubstitution: reference.conf @ jar:file:flink-app-1.0-SNAPSHOT-jar-with-dependencies.jar!/reference.conf: 875: Could not resolve substitution to a value: ${akka.stream.materializer}

  解决：报错：Flink Could not resolve substitution to a value: ${akka.stream.materializer}

    <build><plugins><!-- Java Compiler --><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.1</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>2.3</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><transformers><!--<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"><mainClass>flink.KafkaDemo1</mainClass></transformer>--><transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer"><resource>reference.conf</resource></transformer></transformers></configuration></execution></executions></plugin></plugins></build>

三、样例三：DataSet、Table Sql 处理后写入 StarRocks

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.types.Row;public class SQLWordCount {public static void main(String[] args) throws Exception {TypeInformation[] fieldTypes = {Types.STRING, Types.INT};RowTypeInfo rowTypeInfo = new RowTypeInfo(fieldTypes);JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat().setDrivername("com.mysql.jdbc.Driver").setDBUrl("jdbc:mysql://192.168.xx.xx:9030/dwd?characterEncoding=utf8").setUsername("root").setPassword("").setQuery("select * from student").setRowTypeInfo(rowTypeInfo).finish();ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();BatchTableEnvironment bTableEnv = BatchTableEnvironment.create(env);DataSet<Row> dataSource = env.createInput(jdbcInputFormat);dataSource.print();Table students = bTableEnv.fromDataSet(dataSource);bTableEnv.registerTable("student", students);Table result = bTableEnv.sqlQuery("select name, age from (select f0 as name, f1 as age from student) group by name, age");result.printSchema();DataSet<Row> dset = bTableEnv.toDataSet(result, Row.class);dset.output(JDBCOutputFormat.buildJDBCOutputFormat().setDrivername("com.mysql.jdbc.Driver").setDBUrl("jdbc:mysql://192.168.xx.xx:9030/dwd?characterEncoding=utf8").setUsername("root").setPassword("").setQuery("insert into student values(?, ?)").finish());env.execute("SQL-Batch");}
}

四、遇到的坑

  坑1：Bang equal '!=' is not allowed under the current SQL conformance level
  解决：将 sql 中的 != 修改为 <>

  坑2：java.lang.RuntimeException: No new data sinks have been defined since the last execution. The last execution refers to the latest call to 'execute()', 'count()', 'collect()', or 'print()'.
  解释：在最后一行代码 env.execute() 执行的时候，没有新的数据接收器被定义，对于 Flink 批处理而前一行代码 result.print() 已经触发了代码的执行和输出，所以再执行 env.execute()，就是多余的了，因此报了上面的异常。
  解决方法：去掉最后一行代码 env.execute(); 就可以了。

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

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

相关文章

RabbitMQ 应用问题

RabbitMQ 应用问题

文章目录 1. 幂等性保障什么是幂等性MQ 的幂等性如何处理消息重复的问题1. 全局唯一ID2. 业务逻辑判断 2. 顺序性保障什么是顺序性保障什么情况会打破RabbitMQ的顺序性顺序性保障方案 3. 消息积压什么是消息积压造成消息积压的原因解决消息积压的方案 结论 1. 幂等性保障 什么…
阅读更多...
【数据库系列】MySQL基础知识：深入理解DDL、DML与DQL操作

【数据库系列】MySQL基础知识：深入理解DDL、DML与DQL操作

MySQL是一个开源的关系型数据库管理系统&#xff08;RDBMS&#xff09;&#xff0c;广泛用于数据存储和管理。理解MySQL的基本操作至关重要&#xff0c;尤其是数据定义语言&#xff08;DDL&#xff09;、数据操作语言&#xff08;DML&#xff09;和数据查询语言&#xff08;DQL…
阅读更多...
PAT1085 Perfect Sequence(25)

PAT1085 Perfect Sequence(25)

//判断是否是连续的数 //判断是否只能第一个数是最小值 #include <cstdio> #include <algorithm> typedef long long ll; using namespace std; int n,p; const int maxn 100010; int arr[maxn];int binary(int l, int r, ll tgt){if(arr[n-1] < tgt) return n…
阅读更多...
Shell 编程基础知识

Shell 编程基础知识

为什么要学 Shell&#xff1f; 学一个东西&#xff0c;我们大部分情况都是往实用性方向着想。从工作角度来讲&#xff0c;学习 Shell 是为了提高我们自己工作效率&#xff0c;提高产出&#xff0c;让我们在更少的时间完成更多的事情。 很多人会说 Shell 编程属于运维方面的知…
阅读更多...
深入浅出UART驱动开发与调试：从基础调试到虚拟驱动实现

深入浅出UART驱动开发与调试：从基础调试到虚拟驱动实现

往期内容 本专栏往期内容&#xff1a;Uart子系统 UART串口硬件介绍深入理解TTY体系&#xff1a;设备节点与驱动程序框架详解Linux串口应用编程&#xff1a;从UART到GPS模块及字符设备驱动 解UART 子系统&#xff1a;Linux Kernel 4.9.88 中的核心结构体与设计详解IMX 平台UART驱…
阅读更多...
Linux网络——IO模型和多路转接

Linux网络——IO模型和多路转接

通常所谓的IO&#xff0c;其本质就是等待通信和进行通信&#xff0c;即IO 等 拷贝。 那么想要做到高效的IO&#xff0c;就要在单位时间内&#xff0c;减少“等”的比重。 一.五种IO模型 阻塞 IO: 在内核将数据准备好之前, 系统调用会一直等待. 所有的套接字, 默认都是阻塞方…
阅读更多...
VM Virutal Box的Ubuntu虚拟机与windows宿主机之间设置共享文件夹(自动挂载，永久有效）

VM Virutal Box的Ubuntu虚拟机与windows宿主机之间设置共享文件夹(自动挂载，永久有效）

本文参考如下链接 How to access a shared folder in VirtualBox? - Ask Ubuntu &#xff08;1&#xff09;安装增强功能&#xff08;Guest Additions&#xff09; 首先&#xff0c;在网上下载VBoxGuestAdditions光盘映像文件 下载地址&#xff1a;Index of http://…
阅读更多...
AI的魔力：如何为开源软件注入智慧，开启无限可能

AI的魔力：如何为开源软件注入智慧，开启无限可能

“AI的魔力&#xff1a;如何为开源软件注入智慧&#xff0c;开启无限可能” 引言&#xff1a; 在科技发展的浪潮中&#xff0c;开源软件生态一直扮演着推动创新与共享的重要角色。从Linux到Python&#xff0c;开源项目赋予了开发者全球协作的机会&#xff0c;推动了技术的飞速…
阅读更多...
IThenticate 查重有无免费午餐？深度解析

IThenticate 查重有无免费午餐？深度解析

经历过论文“折磨”的过来人&#xff0c;深知查重工具是写论文不可或缺的助手。而 iThenticate 查重系统&#xff0c;深受出版商、学术机构和研究人员喜爱。不过&#xff0c;每次看到它那昂贵的价格&#xff0c;就让很多小伙伴直呼&#xff0c;IThenticate查重系统就没有免费的…
阅读更多...
启动SpringBoot

启动SpringBoot

前言&#xff1a;大家好我是小帅&#xff0c;今天我们来学习SpringBoot 文章目录 1. 环境准备2. Maven2.1 什么是Maven2.2 创建⼀个Maven项⽬2.3 依赖管理2.3.1 依赖配置2.3.2 依赖传递2.3.4 依赖排除2.3.5 Maven Help插件&#xff08;plugin&#xff09; 2.4 Maven 仓库2.6 中…
阅读更多...
DHCP服务（包含配置过程）

DHCP服务（包含配置过程）

目录 一、 DHCP的定义 二、 使用DHCP的好处 三、 DHCP的分配方式 四、 DHCP的租约过程 1. 客户机请求IP 2. 服务器响应 3. 客户机选择IP 4. 服务器确定租约 5. 重新登录 6. 更新租约 五、 DHCP服务配置过程 一、 DHCP的定义 DHCP&#xff08;Dynamic Host Configur…
阅读更多...
使用 Certbot 为 Nginx 自动配置 SSL 证书

使用 Certbot 为 Nginx 自动配置 SSL 证书

1.安装Certbot和Nginx插件 sudo apt-get update sudo apt-get install certbot python3-certbot-nginx 2.获取和安装证书 运行Certbot自动安装SSL证书。注意替换 your_domain sudo certbot --nginx -d your_domain Certbot将自动与Lets Encrypt的服务器通信&#xff0c;验证域…
阅读更多...
ros2键盘实现车辆: 简单的油门_刹车_挡位_前后左右移动控制

ros2键盘实现车辆: 简单的油门_刹车_挡位_前后左右移动控制

参考: ROS python 实现键盘控制 底盘移动 https://blog.csdn.net/u011326325/article/details/131609340游戏手柄控制 1.背景与需求 1.之前实现过 键盘控制 底盘移动的程序, 底盘是线速度控制, 效果还不错. 2.新的底盘 只支持油门控制, 使用线速度控制问题比较多, 和底盘适配…
阅读更多...
DICOM医学影像应用篇——窗宽窗位概念、原理及实现详解

DICOM医学影像应用篇——窗宽窗位概念、原理及实现详解

目录 窗宽窗位调整&#xff08;Windowing&#xff09;在DICOM医学影像中的应用 窗宽窗位的基本概念 窗宽&#xff08;Window Width, WW&#xff09; 窗位&#xff08;Window Level, WL&#xff09; 窗宽窗位调整的基本原理 映射逻辑 数学公式 窗宽窗位调整的C实现 代码…
阅读更多...
天锐绿盾加密软件与Ping32联合打造企业级安全保护系统，确保敏感数据防泄密与加密管理

天锐绿盾加密软件与Ping32联合打造企业级安全保护系统，确保敏感数据防泄密与加密管理

随着信息技术的飞速发展&#xff0c;企业在日常经营过程中产生和处理的大量敏感数据&#xff0c;面临着越来越复杂的安全威胁。尤其是在金融、医疗、法律等领域&#xff0c;数据泄漏不仅会造成企业巨大的经济损失&#xff0c;还可能破坏企业的信誉和客户信任。因此&#xff0c;…
阅读更多...
HarmonyOS：@Provide装饰器和@Consume装饰器：与后代组件双向同步

HarmonyOS：@Provide装饰器和@Consume装饰器：与后代组件双向同步

一、前言 Provide和Consume&#xff0c;应用于与后代组件的双向数据同步&#xff0c;应用于状态数据在多个层级之间传递的场景。不同于上文提到的父子组件之间通过命名参数机制传递&#xff0c;Provide和Consume摆脱参数传递机制的束缚&#xff0c;实现跨层级传递。 其中Provi…
阅读更多...
【Spring MVC】如何运用应用分层思想实现简单图书管理系统前后端交互工作

【Spring MVC】如何运用应用分层思想实现简单图书管理系统前后端交互工作

前言 &#x1f31f;&#x1f31f;本期讲解关于SpringMVC的编程思想之应用分层~~~ &#x1f308;感兴趣的小伙伴看一看小编主页&#xff1a;GGBondlctrl-CSDN博客 &#x1f525; 你的点赞就是小编不断更新的最大动力 &#x1f386;那…
阅读更多...
【Linux】项目自动化构建工具-make/Makefile

【Linux】项目自动化构建工具-make/Makefile

【Linux】项目自动化构建工具-make/Makefile make 和 makefile 的概念如何清理项目推导过程Linux第⼀个小程序−倒计时 &#x1f30f;个人博客主页&#xff1a;个人主页 make 和 makefile 的概念 make是一个命令工具&#xff0c;是一个解释makefile中指令的命令工具&#xf…
阅读更多...
arcgis for js点击聚合要素查询其包含的所有要素

arcgis for js点击聚合要素查询其包含的所有要素

功能说明 上一篇讲了实现聚合效果, 但是点击聚合效果无法获取到该聚合点包含的所有点信息 这一篇是对如何实现该功能的案例 实现 各属性说明需要自行去官网查阅 官网案例 聚合API 没空说废话了, 加班到12点,得休息了, 直接运行代码看效果就行, 相关重点和注意事项都在代码注…
阅读更多...
【计算机视觉】图像基本操作

【计算机视觉】图像基本操作

1. 数字图像表示 一幅尺寸为MN的图像可以用矩阵表示&#xff0c;每个矩阵元素代表一个像素&#xff0c;元素的值代表这个位置图像的亮度&#xff1b;其中&#xff0c;彩色图像使用3维矩阵MN3表示&#xff1b;对于图像显示来说&#xff0c;一般使用无符号8位整数来表示图像亮度&…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023