devbox
知新_RL
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink 读写MySQL数据（DataStream和Table API）_flink mysql

作者：知新_RL | 2024-08-20 04:30:50

赞

踩

flink mysql

Flink提供了基于JDBC的方式，可以将读取到的数据写入到MySQL中；本文通过两种方式将数据下入到MySQL数据库，其他的基于JDBC的数据库类似，另外，Table API方式的Catalog指定为Hive Catalog方式，持久化DDL操作。

另外，JDBC 连接器允许使用 JDBC 驱动程序从任何关系数据库读取数据并将数据写入其中。 本文档介绍如何设置 JDBC 连接器以针对关系数据库运行 SQL 查询。

如果 DDL 上定义了主键，则 JDBC sink 以 upsert 模式与外部系统交换 UPDATE/DELETE 消息，否则，它以 append 模式运行，不支持消费 UPDATE/DELETE 消息。

默认提供 exactly-once的保证。

MySQL配置

MySQL中创建表Events数据表

  1. -- flink.events definition
  2.  
  3. CREATE TABLE `events` (
  4.   `user` varchar(100) DEFAULT NULL,
  5.   `url` varchar(200) DEFAULT NULL,
  6.   `timestamp` bigint(20) DEFAULT NULL
  7. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

插入几条数据

  1. INSERT INTO flink.events
  2. (`user`, url, `timestamp`)
  3. VALUES('Alice', './home', 1000);

Maven依赖，包含了Hive Catalog的相关依赖
 

  1.  
  2.  
  3.     <properties>
  4.         <maven.compiler.source>8</maven.compiler.source>
  5.         <maven.compiler.target>8</maven.compiler.target>
  6.         <flink.version>1.14.4</flink.version>
  7.         <scala.binary.version>2.12</scala.binary.version>
  8.         <hadoop.version>3.1.2</hadoop.version>
  9.         <hive.version>3.1.2</hive.version>
  10.     </properties>
  11.  
  12.  
  13.     <dependencies>
  14.  
  15.         <dependency>
  16.             <groupId>org.apache.flink</groupId>
  17.             <artifactId>flink-java</artifactId>
  18.             <version>${flink.version}</version>
  19.         </dependency>
  20.  
  21.         <dependency>
  22.             <groupId>org.apache.flink</groupId>
  23.             <artifactId>flink-streaming-java_2.12</artifactId>
  24.             <version>${flink.version}</version>
  25.             <!--            <scope>provided</scope>-->
  26.         </dependency>
  27.         <dependency>
  28.             <groupId>org.apache.flink</groupId>
  29.             <artifactId>flink-clients_2.12</artifactId>
  30.             <version>${flink.version}</version>
  31.         </dependency>
  32.  
  33.         <dependency>
  34.             <groupId>org.apache.flink</groupId>
  35.             <artifactId>flink-table-api-java</artifactId>
  36.             <version>${flink.version}</version>
  37.         </dependency>
  38.  
  39.         <dependency>
  40.             <groupId>org.apache.flink</groupId>
  41.             <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
  42.             <version>${flink.version}</version>
  43.         </dependency>
  44.  
  45.  
  46.         <dependency>
  47.             <groupId>mysql</groupId>
  48.             <artifactId>mysql-connector-java</artifactId>
  49.             <version>5.1.49</version>
  50.         </dependency>
  51.  
  52.         <dependency>
  53.             <groupId>org.apache.flink</groupId>
  54.             <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
  55.             <version>${flink.version}</version>
  56.         </dependency>
  57.  
  58.  
  59.         <!-- Flink 的 Hive 连接器-->
  60.         <dependency>
  61.             <groupId>org.apache.flink</groupId>
  62.             <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
  63.             <version>${flink.version}</version>
  64.         </dependency>
  65.         <!-- Hive 依赖 -->
  66.         <dependency>
  67.             <groupId>org.apache.hive</groupId>
  68.             <artifactId>hive-exec</artifactId>
  69.             <version>${hive.version}</version>
  70.             <exclusions>
  71.                 <exclusion>
  72.                     <groupId>org.apache.calcite</groupId>
  73.                     <artifactId>*</artifactId>
  74.                 </exclusion>
  75.             </exclusions>
  76.         </dependency>
  77.         <!-- Hive 依赖 -->
  78.         <dependency>
  79.             <groupId>org.apache.hadoop</groupId>
  80.             <artifactId>hadoop-common</artifactId>
  81.             <version>${hadoop.version}</version>
  82.         </dependency>
  83.         <!-- Hive 依赖 -->
  84.         <dependency>
  85.             <groupId>org.apache.hadoop</groupId>
  86.             <artifactId>hadoop-mapreduce-client-core</artifactId>
  87.             <version>${hadoop.version}</version>
  88.         </dependency>
  89.         <!-- Hive 依赖 -->
  90.         <dependency>
  91.             <groupId>org.apache.hadoop</groupId>
  92.             <artifactId>hadoop-hdfs</artifactId>
  93.             <version>${hadoop.version}</version>
  94.         </dependency>
  95.  
  96.  
  97.     </dependencies>

DataStream方式读写MySQL数据

  1. import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
  2. import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
  3. import org.apache.flink.connector.jdbc.JdbcSink;
  4. import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7.  
  8. import java.sql.PreparedStatement;
  9. import java.sql.SQLException;
  10.  
  11. public class JdbcSinkExample {
  12.  
  13.     public static void main(String[] args) throws Exception {
  14.  
  15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.         env.setParallelism(1);
  17.         SingleOutputStreamOperator<Event> eventStream = env.fromElements(
  18.                 new Event("Alice", "./home", 1000L),
  19.                 new Event("Bob", "./cart", 2000L),
  20.                 new Event("Alice", "./prod?id=1", 5 * 1000L),
  21.                 new Event("Cary", "./home", 60 * 1000L),
  22.                 new Event("Bob", "./prod?id=3", 90 * 1000L),
  23.                 new Event("Alice", "./prod?id=1", 105 * 1000L)
  24.         );
  25.  
  26.  
  27.         eventStream.addSink(JdbcSink.sink(
  28.                 "insert into events (user, url, timestamp) values (?, ?, ?)",
  29.                 new JdbcStatementBuilder<Event>() {
  30.                     @Override
  31.                     public void accept(PreparedStatement preparedStatement, Event event) throws SQLException {
  32.                         preparedStatement.setString(1, event.user);
  33.                         preparedStatement.setString(2, event.url);
  34.                         preparedStatement.setLong(3, event.timestamp);
  35.                     }
  36.                 },
  37.                 JdbcExecutionOptions.builder()
  38.                         .withBatchSize(1000)
  39.                         .withBatchIntervalMs(200)
  40.                         .withMaxRetries(5)
  41.                         .build(),
  42.                 new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
  43.                         .withUrl("jdbc:mysql://127.0.0.1:3306/flink")
  44.                         .withDriverName("com.mysql.jdbc.Driver")
  45.                         .withUsername("root")
  46.                         .withPassword("xxx")
  47.                         .build()
  48.         ));
  49.  
  50.         env.execute();
  51.     }
  52. }

Table API的方式读写MySQL，其中Flink的Catalog使用Hive Catalog的方式。熟悉 Flink 或者 Spark 等大数据引擎的同学应该都知道这两个计算引擎都有一个共同的组件叫 Catalog。下面是 Flink 的 Catalog 的官方定义。

Catalog 提供了元数据信息，例如数据库、表、分区、视图以及数据库或其他外部系统中存储的函数和信息。

Table API与SQL API实现了Apache Flink的批流统一的实现方式。Table API与SQL API的核心概念就是TableEnviroment。TableEnviroment对象提供方法注册数据源与数据表信息。那么数据源与数据表的信息则存储在CataLog中。所以，CataLog是TableEnviroment的重要组成部分。”

Apache Flink在获取TableEnviroment对象后，可以通过Register实现对数据源与数据表进行注册。注册完成后数据库与数据表的原信息则存储在CataLog中。CataLog中保存了所有的表结构信息、数据目录信息等。

简单来说，Catalog 就是元数据管理中心，其中元数据包括数据库、表、表结构等信息。

  1. public class JDBCTable {
  2.     public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4.         env.setParallelism(1);
  5.         SingleOutputStreamOperator<Event> eventStream = env.fromElements(
  6.                 new Event("Alice", "./home", 1000L),
  7.                 new Event("Bob", "./cart", 2000L),
  8.                 new Event("Alice", "./prod?id=1", 5 * 1000L),
  9.                 new Event("Cary", "./home", 60 * 1000L),
  10.                 new Event("Bob", "./prod?id=3", 90 * 1000L),
  11.                 new Event("Alice", "./prod?id=1", 105 * 1000L)
  12.         );
  13.  
  14.         //获取表环境
  15.         //StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
  16.         EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().build();
  17.         TableEnvironment tableEnv = TableEnvironment.create(settings);
  18.  
  19.         String name = "myhive";
  20.         String defaultDatabase = "default";
  21.         String hiveConfDir = "/opt/hive";
  22.  
  23.         // 创建一个 HiveCatalog，并在表环境中注册
  24.         HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
  25.         tableEnv.registerCatalog("myhive", hive);
  26.  
  27.         // 使用 HiveCatalog 作为当前会话的 catalog
  28.         tableEnv.useCatalog("myhive");
  29.  
  30.  
  31.         TableResult tableResult = tableEnv.executeSql("CREATE TABLE IF NOT EXISTS EventTable (\n" +
  32.                 "`user` STRING,\n" +
  33.                 "url STRING,\n" +
  34.                 "`timestamp` BIGINT\n" +
  35.                 ") WITH (\n" +
  36.                 "'connector' = 'jdbc',\n" +
  37.                 "'url' = 'jdbc:mysql://127.0.0.1:3306/flink',\n" +
  38.                 "'table-name' = 'events',\n" +
  39.                 "'username'='root',\n" +
  40.                 "'password'='xxx'\n" +
  41.                 ")");
  42.         tableEnv.executeSql("insert into EventTable values('Alice','./prod?id=3',106000)");
  43.     }

MySQL中的数据

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/知新_RL/article/detail/1005205
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号