大数据-117 - Flink DataStream Sink 案例：写出到MySQL、写出到Kafka ...

点一下关注吧！！！非常感谢！！连续更新！！！

现在已经更新到了：

• Hadoop（已更完）
  
• HDFS（已更完）
  
• MapReduce（已更完）
  
• Hive（已更完）
  
• Flume（已更完）
  
• Sqoop（已更完）
  
• Zookeeper（已更完）
  
• HBase（已更完）
  
• Redis （已更完）
  
• Kafka（已更完）
  
• Spark（已更完）
  
• Flink（正在更新！）
  

章节内容

上节我们完成了如下的内容：

• Sink 的基本概念等内容
  
• Sink的相关信息 配置与利用
  
• Sink案例写入Redis
  

JDBC Sink

在 Apache Flink 中，通过 JDBC Sink，可以将处理后的数据写入到 MySQL 数据库中。这对于将实时处理的数据持久化或与其他系统进行集成非常有用。
Flink JDBC Sink 简介

Flink 提供了 JdbcSink，它是基于 JDBC 协议的 Sink，可以将数据写入各种关系型数据库，包罗 MySQL。在利用 JDBC Sink 时，必要提供数据库毗连信息和 SQL 语句，通过这些信息，Flink 将数据流中的纪录插入或更新到 MySQL 表中。
Flink 到 MySQL 的基本步调

将数据流写入 MySQL 的步调主要包罗以下几点：

• 依赖库配置：确保在项目中引入了 Flink 和 MySQL 相关的依赖库，通常必要配置 Maven 或 Gradle。
  
• 界说数据源和数据流：创建并处理数据流。
  
• 配置 JDBC Sink：提供数据库的毗连信息和插入 SQL 语句。
  
• 启动使命：将数据流写入 MySQL。
  

优化建议

在实际项目中，向 MySQL 插入大量数据时，应考虑以下优化策略：

• 批量插入：通过 JdbcExecutionOptions 配置批量插入，可以大幅提拔写入性能。
  
• 毗连池：对于高并发的写入操作，建议利用毗连池来减少数据库毗连开销。
  
• 索引优化：为插入的表配置合适的索引，可以提高查询性能，但在大量写入时，索引可能会降低- 插入速率，因此必要权衡。
  
• 数据分片：对于非常大规模的数据，可以考虑将数据分片并行写入不同的 MySQL 实例或分区表中。
  

案例：流数据下沉到MySQL

添加依赖

1. <dependency>
2.     <groupId>mysql</groupId>
3.     <artifactId>mysql-connector-java</artifactId>
4.     <version>8.0.28</version>
5. </dependency>

复制代码

编写代码

一个Person的类，对应MySQL中的一张表的字段。
模拟几条数据流，写入到 MySQL中。

1. package icu.wzk;
4. import org.apache.flink.configuration.Configuration;
5. import org.apache.flink.streaming.api.datastream.DataStreamSource;
6. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
7. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;
9. import java.sql.Connection;
10. import java.sql.DriverManager;
11. import java.sql.PreparedStatement;
13. public class SinkSqlTest {
15.     public static void main(String[] args) {
16.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
17.         DataStreamSource<Person> data = env.getJavaEnv().fromElements(
18.                 new Person("wzk", 18, 1),
19.                 new Person("icu", 20, 1),
20.                 new Person("wzkicu", 13, 2)
21.         );
22.         data.addSink(new MySqlSinkFunction());
24.         env.execute();
25.     }
27.     public static class MySqlSinkFunction extends RichSinkFunction<Person> {
29.         private PreparedStatement preparedStatement = null;
31.         private Connection connection = null;
33.         @Override
34.         public void open(Configuration parameters) throws Exception {
35.             String url = "jdbc:mysql://h122.wzk.icu:3306/flink-test?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC";
36.             String username = "hive";
37.             String password = "hive@wzk.icu";
38.             connection = DriverManager.getConnection(url, username, password);
39.             String sql = "INSERT INTI PERSON(name, age, sex) VALUES(?, ?, ?)";
40.             preparedStatement = connection.prepareStatement(sql);
41.         }
43.         @Override
44.         public void invoke(Person value, Context context) throws Exception {
45.             preparedStatement.setString(1, value.getName());
46.             preparedStatement.setInt(2, value.getAge());
47.             preparedStatement.setInt(3, value.getSex());
48.             preparedStatement.executeUpdate();
49.         }
51.         @Override
52.         public void close() throws Exception {
53.             if (null != connection) {
54.                 connection.close();
55.             }
56.             if (null != preparedStatement) {
57.                 preparedStatement.close();
58.             }
59.         }
60.     }
62.     public static class Person {
63.         private String name;
64.         private Integer age;
65.         private Integer sex;
67.         public Person() {
69.         }
71.         public Person(String name, Integer age, Integer sex) {
72.             this.name = name;
73.             this.age = age;
74.             this.sex = sex;
75.         }
77.         public String getName() {
78.             return name;
79.         }
81.         public void setName(String name) {
82.             this.name = name;
83.         }
85.         public Integer getAge() {
86.             return age;
87.         }
89.         public void setAge(Integer age) {
90.             this.age = age;
91.         }
93.         public Integer getSex() {
94.             return sex;
95.         }
97.         public void setSex(Integer sex) {
98.             this.sex = sex;
99.         }
100.     }
101. }

复制代码

数据库配置

我们新建一张表出来，person表，里边有我们必要的字段。

运行代码

我们运行代码，等候运行结束。

查察结果

查察数据库中的数据，我们可以看到刚才模拟的数据已经成功写入了。

案例：写入到Kafka

编写代码

1. package icu.wzk;
3. import org.apache.flink.api.common.serialization.SimpleStringSchema;
4. import org.apache.flink.streaming.api.scala.DataStream;
5. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;
6. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
8. public class SinkKafkaTest {
10.     public static void main(String[] args) {
11.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
12.         DataStream<String> data = env.socketTextStream("localhost", 9999, '\n', 0);
13.         String brokerList = "h121.wzk.icu:9092";
14.         String topic = "flink_test";
15.         FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(brokerList, topic, new SimpleStringSchema());
16.         data.addSink(producer);
17.         env.execute("SinkKafkaTest");
18.     }
20. }

复制代码

运行代码

启动一个 nc

1. nc -lk 9999

复制代码

我们通过回车的方式，可以发送数据。

Java 步伐中等候

查察结果

我们登录到服务器查察信息

1. ./kafka-console-consumer.sh --bootstrap-server h121.wzk.icu:9092 --topic flink_test --from-beginning

复制代码

可以看到刚才的数据已经写入了：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。