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、加载数据源到 DataSet
        DataSet<Student> csv = env.readCsvFile("D:\\tmp\\data.csv").ignoreFirstLine().pojoType(Student.class, "name", "age");

        // 3、将DataSet装换为Table
        Table 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");
    }

    @Data
    public 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(); 就可以了。

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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

Django+Nginx+uwsgi网站Channels+redis+daphne多人在线聊天实现粘贴上传图片

Django+Nginx+uwsgi网站Channels+redis+daphne多人在线聊天实现粘贴上传图片

在DjangoNginxuwsgi网站Channelsredisdaphne多人在线的基础上&#xff08;详见DjangoNginxuwsgi网站使用Channelsredisdaphne实现简单的多人在线聊天及消息存储功能-CSDN博客&#xff09;&#xff0c;实现在输入框粘贴或打开本地图片&#xff0c;上传到网站后返回图片路径&…
阅读更多...
【Git】Git 命令参考手册

【Git】Git 命令参考手册

目录 Git 命令参考手册1. 创建仓库1.1 创建一个新的本地仓库1.2 克隆一个仓库1.3 克隆仓库到指定目录 2. 提交更改2.1 显示工作目录中已修改的文件&#xff0c;准备提交2.2 将文件添加到暂存区&#xff0c;准备提交2.3 将所有已修改的文件添加到暂存区&#xff0c;准备提交2.4 …
阅读更多...
TDengine在debian安装

TDengine在debian安装

参考官网文档&#xff1a; 官网安装文档链接 从列表中下载获得 Deb 安装包&#xff1b; TDengine-server-3.3.4.3-Linux-x64.deb (61 M) 进入到安装包所在目录&#xff0c;执行如下的安装命令&#xff1a; sudo dpkg -i TDengine-server-<version>-Linux-x64.debNOTE 当…
阅读更多...
linux安装mysql8.0.40

linux安装mysql8.0.40

一、下载MySQL安装包 1.查看glibc版本 rpm -qa | grep glibc 2.到mysql官网下载安装包 ​ 二、解压安装 1.上传压缩包纸/usr/local 目录下&#xff0c;解压&#xff1a; tar -xvf mysql-8.0.40-linux-glibc2.17-x86_64.tar.xz 2.重命名&#xff1a; mv mysql-8.0.40-linux-…
阅读更多...
用Pycharm安装manim

用Pycharm安装manim

由于版本和工具的差异&#xff0c;manim的安装方式不尽相同。本文用Pycharm来安装manim. 一、准备工作&#xff1a;安装相应版本的python、pycharm和ffmpeg. 此处提供一种安装ffmpeg的方式 下载地址&#xff1a;FFmpeg 下载后&#xff0c;解压到指定目录。 配置环境变量&am…
阅读更多...
运维面试整理总结

运维面试整理总结

面试题可以参考:面试题总结 查看系统相关信息 查看系统登陆成功与失败记录 成功&#xff1a;last失败&#xff1a;lastb 查看二进制文件 hexdump查看进程端口或连接 netstat -nltp ss -nltp补充&#xff1a;pidof与lsof命令 pidof [进程名] #根据 进程名 查询进程id ls…
阅读更多...
Kubernetes 之 Ingress 和 Service 的异同点

Kubernetes 之 Ingress 和 Service 的异同点

1. 概念与作用 1.1 Ingress Ingress 是什么&#xff1f; Ingress主要负责七层负载&#xff0c;将外部 HTTP/HTTPS 请求路由到集群内部的服务。它可以基于域名和路径定义规则&#xff0c;从而将外部请求分配到不同的服务。 ingress作用 提供 基于 HTTP/HTTPS 的路由。 支持 …
阅读更多...
初识 Django

初识 Django

声明 适用于想要快速入门的开发者&#xff0c;有前后端开发以及语言基础&#xff0c;想要学习语法或者特性。 想要学会快速开发&#xff0c;快速入门&#xff0c;请看博客【实用向】Django 框架入门并结合本篇文章。 命令 命令描述startproject创建一个 Django 项目startapp…
阅读更多...
python简单算法

python简单算法

冒泡 def boll(lis):i 0while i<len(lis)-1:j 0while j<len(lis)-1-i:if lis[j] > lis[j1]:lis[j],lis[j 1] lis[j1],lis[j]j1i1选择排序 def selct1(lit):i 0while i<len(lit)-1:j i1min1 iwhile j < len(lit):if lit[j] < lit[min1]:min1 jj 1li…
阅读更多...
【大模型】基于LLaMA-Factory的模型高效微调

【大模型】基于LLaMA-Factory的模型高效微调

LLaMA-Factory项目介绍 LLaMA Factory 是一个简单易用且高效的大型语言模型&#xff08;Large Language Model&#xff09;训练与微调平台。通过 LLaMA Factory&#xff0c;可以在无需编写任何代码的前提下&#xff0c;在本地完成上百种预训练模型的微调&#xff0c;框架特性包…
阅读更多...
电脑中的vcruntime140_1.dll文件有问题要怎么解决？一键修复vcruntime140_1.dll

电脑中的vcruntime140_1.dll文件有问题要怎么解决？一键修复vcruntime140_1.dll

遇到“vcruntime140_1.dll无法继续执行代码”的错误通常表明电脑中的vcruntime140_1.dll文件有问题。这个文件属于Visual C Redistributable&#xff0c;对很多程序的运行至关重要。本文将提供几个步骤&#xff0c;帮助你迅速修复这一错误&#xff0c;使电脑恢复正常工作状态。…
阅读更多...
鼠标前进后退键改双击，键盘映射(AutoHotkey)

鼠标前进后退键改双击，键盘映射(AutoHotkey)

初衷&#xff1a; 1.大部分鼠标为不可自定义按键&#xff0c;可以自定义的又很贵。 鼠标左键是双击是很频类很高的操作&#xff0c;鼠标前进/后退按键个人感觉使用频率很低&#xff0c;因此把鼠标前进/后退改为双击还是很合适的。 2.有些短款的键盘没有Home或End键&#xff0c;…
阅读更多...
VSCode Terminal无法运行node以及node-gyp等指令

VSCode Terminal无法运行node以及node-gyp等指令

无法使用node指令&#xff0c;使用管理员权限启动VSCode即可&#xff0c;或者右键VSCode属性&#xff0c;修改兼容性中使用管理员权限打开。 运行node-gyp等指令出现因为在此系统上禁止运行脚本。有关详细信息&#xff0c;请参阅 https:/go.microsoft.com/fwlink/?LinkID1351…
阅读更多...
npm install -g@vue/cli报错解决：npm error code ENOENT npm error syscall open

npm install -g@vue/cli报错解决：npm error code ENOENT npm error syscall open

这里写目录标题 报错信息1解决方案 报错信息2解决方案 报错信息1 使用npm install -gvue/cli时&#xff0c;发生报错&#xff0c;报错图片如下&#xff1a; 根据报错信息可以知道&#xff0c;缺少package.json文件。 解决方案 缺什么补什么&#xff0c;这里我们使用命令npm…
阅读更多...
Elasticsearch：Retrievers 介绍

Elasticsearch：Retrievers 介绍

检索器&#xff08;retrievers&#xff09;是 Elasticsearch 中搜索 API 中添加的新抽象层。它们提供了在单个 _search API 调用中配置多阶段检索管道的便利。此架构通过消除对复杂搜索查询的多个 Elasticsearch API 调用的需求&#xff0c;简化了应用程序中的搜索逻辑。它还减…
阅读更多...
nvidia-container-toolkit安装问题（OpenPGP）

nvidia-container-toolkit安装问题（OpenPGP）

1.正常情况下 apt-get install -y nvidia-container-toolkit2.使用nvidia源 nvidia-container-toolkit官网有安装教程 2.1 配置生产存储库 curl -fsSL https://nvidia.github.io/libnvidia-container/gpgkey | sudo gpg --dearmor -o /usr/share/keyrings/nvidia-containe…
阅读更多...
open-instruct - 训练开放式指令跟随语言模型

open-instruct - 训练开放式指令跟随语言模型

文章目录 关于 open-instruct设置训练微调偏好调整RLVR 污染检查开发中仓库结构 致谢 关于 open-instruct github : https://github.com/allenai/open-instruct 这个仓库是我们对在公共数据集上对流行的预训练语言模型进行指令微调的开放努力。我们发布这个仓库&#xff0c;并…
阅读更多...
38 基于单片机的宠物喂食(ESP8266、红外、电机)

38 基于单片机的宠物喂食(ESP8266、红外、电机)

目录 一、主要功能 二、硬件资源 三、程序编程 四、实现现象 一、主要功能 基于STC89C52单片机&#xff0c;采用L298N驱动连接P2.3和P2.4口进行电机驱动&#xff0c; 然后串口连接P3.0和P3.1模拟ESP8266&#xff0c; 红外传感器连接ADC0832数模转换器连接单片机的P1.0~P1.…
阅读更多...
Mac安装及合规无限使用Beyond Compare

Mac安装及合规无限使用Beyond Compare

文章目录 Beyond CompareBeyond Compare简介Beyond Compare安装Beyond Compare到期后继续免费使用 Beyond Compare Beyond Compare简介 Beyond Compare 是一款由 Scooter Software 开发的文件和文件夹比较工具。它主要用于对比两个文件或文件夹之间的差异&#xff0c;并支持文…
阅读更多...
详细分析 npm run build 基本知识 | 不同环境不同命令

详细分析 npm run build 基本知识 | 不同环境不同命令

目录 前言1. 基本知识2. 构建逻辑 前言 关于部署服务器的知识推荐阅读&#xff1a;npm run build部署到云服务器中的Nginx&#xff08;图文配置&#xff09; 1. 基本知识 npm run 是 npm 的一个命令&#xff0c;用于运行 package.json 中定义的脚本&#xff0c;可以通过 “s…
阅读更多...
最新文章

Copyright @ 2022~2023