Flink基础入门一

作为一个系列主要从flink入门开始到深入 flink作为流式处理框架这些简单介绍可以直接查看官网

基础案例

1. pom文件导入

    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <flink.version>1.17.1</flink.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
    </dependencies>

2. source数据源

flink的数据源很丰富，这里作为入门先做简单来源数据接下来是简单案例

有使用mysql 导入新的jar

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


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

public class Flink01_Source {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //直接从数据中读取数据
        //DataStreamSource<Integer> envStream = env.fromElements(1, 2, 3, 4, 5, 6);
        //集合中获取数据
        //DataStreamSource<Integer> envStream = env.fromCollection(List.of(1,2,3,4,5,6));
        //直接由socket数据传输获取 nc -l 9090
        //DataStreamSource<String> envStream = env.socketTextStream("localhost", 9090);
        //直接读取文件，但是已过时
        //DataStreamSource<String> envStream = env.readTextFile("data");
        //经常使用的读取数据源mysql 或 kafka
        //简单方案 可以使用flink-cdc方案更加便捷
        DataStreamSource<String> envStream = env.addSource(new RichSourceFunction<String>() {
            private Connection connection;
            private Statement statement;
            private boolean running = true;

            @Override
            public void open(Configuration parameters) throws Exception {
                //创建mysql链接
                Class.forName("com.mysql.cj.jdbc.Driver");
                String url = "jdbc:mysql://localhost:3306/test";
                String username = "root";
                String password = "root@1q2w3e";
                connection = DriverManager.getConnection(url, username, password);
                statement = connection.createStatement();
            }

            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                while (running) {
                    ResultSet rs = statement.executeQuery("SELECT * FROM demo01");
                    while (rs.next()) {
                        // 从结果集中读取数据并发送到Flink流
                        String data = rs.getString("id");
                        ctx.collect(data);
                    }
                    Thread.sleep(1000); // 可以根据需要调整读取间隔
                }
            }

            @Override
            public void cancel() {
                running = false;
                try {
                    if (statement != null) {
                        statement.close();
                    }
                    if (connection != null) {
                        connection.close();
                    }
                } catch (Exception e) {
                    // 处理关闭连接时的异常
                }
            }
        });

        envStream.print();
        env.execute();
    }
}