flink任务编程-java版-流-数据库to数据库

本文就flink流计算的任务工程模板做一介绍，初始数据源和计算结果数据存储都是通过JDBC连接到数据库。本文以mysql为数据库进行演示。

依赖包

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.15</version>
        </dependency>

工程结构说明

工程结构

本工程包含以下几个部分：

1. 结构化数据定义（DTO），本文以常见的学生基本信息为示例。

package com.crazyice.lee.data;

import lombok.*;

@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString(exclude = {"password","age"})
@RequiredArgsConstructor()
public class Student {
    private int studentId;
    private String name;
    private String password;
    private boolean sex;
    private int age;
}

2. 任务编排，这里演示了一个完整流任务的各个关键环节，包括：环境配置、源数据读取（DataStreamSource）、处理过程（数据过滤筛选）、处理结果输出入库（DataSink）、启动任务执行等。

package com.crazyice.lee.jobs;

import com.crazyice.lee.data.Student;
import com.crazyice.lee.reader.JdbcReader;
import com.crazyice.lee.writer.JdbcWriter;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

@Slf4j
public class Mysql2MysqlJob {

    public static void main(String[] args) throws Exception {
        //设置环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //获取数据-数据源
        DataStreamSource<Student> students = env.addSource(new JdbcReader());
        students.name("从mysql读取数据");

        //处理数据过程
        DataStream<Student> subStudents = students
                .filter(e -> e.isSex()).name("过滤男生")
                .filter(e -> e.getAge() > 12).name("过滤12岁以上");

        //写入数据-处理结果
        subStudents.addSink(new JdbcWriter()).name("写入备份学生库");
        env.execute("读取Mysql，数据过滤，写入Mysql");
    }
}

3. 初始数据源、结果写入数据源配置，这部分通过实现RichSourceFunction、RichSinkFunction接口并配合配置文件实现，是相对固定的模式，如果要提高编程效率，可以考虑使用模板及配置的方式动态实现，因为代码结构固定，所以不在此处粘贴，如果感兴趣可以通过文章下面的开源链接查看。

测试验证

flink支持本地测试模式，可以在编码过程中排查错误，只需要在环境设置部分使用本地环境对象即可（将原先的getExecutionEnvironment()替换为createLocalEnvironment()）也可以使用profile来进行编译配置。

        //设置环境
        final ExecutionEnvironment environment = properties.getProperty("local.running").equalsIgnoreCase("true") ? StreamExecutionEnvironment.createLocalEnvironment() : StreamExecutionEnvironment.getExecutionEnvironment();

源代码：（https://gitee.com/crazyicelee/flink）