Flink数据流即时写入MySQL：从头到尾实战

后端

2023-09-24 06:35:55

实时数据集成：使用 Flink 将 Kafka 数据写入 MySQL

简介

在现代数据驱动的世界中，实时数据集成至关重要。它使组织能够分析和利用实时数据，做出明智的决策并响应动态业务环境。本文将指导您使用 Apache Flink 将数据从 Apache Kafka 实时写入 MySQL 数据库。

准备工作

在继续之前，请确保安装了以下软件：

Apache Flink
Apache Kafka
MySQL

此外，创建一个 MySQL 数据库和一个表以存储数据。

连接 MySQL

要从 Flink 流式传输数据到 MySQL，我们需要首先创建一个与数据库的连接。

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;

public class MySQLConnector {

    private static Connection connection;

    public static Connection getConnection() throws SQLException {
        if (connection == null) {
            connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/flink_test", "root", "password");
        }
        return connection;
    }

}

创建 Flink 环境

下一步是设置 Flink 流式传输环境。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkEnvironment {

    private static StreamExecutionEnvironment env;

    public static StreamExecutionEnvironment getEnv() {
        if (env == null) {
            env = StreamExecutionEnvironment.getExecutionEnvironment();
        }
        return env;
    }

}

从 Kafka 摄取数据

从 Kafka 摄取数据需要一个消费者。

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.api.datastream.DataStream;

public class KafkaSource {

    public static DataStream<String> createSource(String topic, String groupId) {
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), PropertiesUtil.getKafkaProperties(groupId));
        return FlinkEnvironment.getEnv().addSource(consumer);
    }

}

写入 MySQL

最后，使用 JDBC Sink 将数据写入 MySQL。

import org.apache.flink.streaming.connectors.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;

public class MySQLSink {

    public static void createSink(DataStream<String> dataStream) {
        JdbcSink<String> sink = JdbcSink.sink(
                "INSERT INTO flink_test.test_table (value) VALUES (?)",
                (statement, value) -> statement.setString(1, value),
                MySQLConnector::getConnection
        );
        dataStream.addSink(sink);
    }

}