flink 1.10 source方式读取mysql

小鱼儿 2022-12-19 08:51 102阅读 0赞

class mysql_source extends RichSourceFunction[(Long, String, Long)] { 
      private var connection: Connection = null
      private var ps: PreparedStatement = null
    
      override def open(parameters: Configuration): Unit = { 
        val driver = "com.mysql.jdbc.Driver"
        val url = "jdbc:mysql://localhost:3306/story"
        val username = "root"
        val password = "root"
        Class.forName(driver)
        connection = DriverManager.getConnection(url, username, password)
        val sql = "select id, name,age from user"
        ps = connection.prepareStatement(sql)
      }
    
      override def close(): Unit = { 
        if (connection != null) { 
          connection.close()
        }
        if (ps != null) { 
          ps.close()
        }
      }
    
    
      override def run(sourceContext: SourceContext[(Long, String, Long)]): Unit = { 
        val queryRequest = ps.executeQuery()
        while (queryRequest.next()) { 
          val id = queryRequest.getLong("id")
          val name = queryRequest.getString("name")
          val age = queryRequest.getLong("age")
          sourceContext.collect((id, name, age))
        }
      }
    
      override def cancel(): Unit = { }
    }

val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val mysqlTable: DataStream[(Long, String, Long)] = env.addSource(new mysql_source())

发表评论取消回复

表情：

评论列表（有 0 条评论，102人围观）

还没有评论，来说两句吧...

相关阅读

相关 Flink案例——kafka、MySQL source

Flink案例——kafka、MySQL source 一、kafka source flink和kafka的连接是十分友好的，毕竟是做流式处理的吧。首先依赖

川长思鸟来/ 2023年10月06日 21:28/ 0 赞/ 10 阅读

相关 flink 1.10 source方式读取mysql

class mysql_source extends RichSourceFunction[(Long, String, Long)] { private

小鱼儿/ 2022年12月19日 08:51/ 0 赞/ 103 阅读

相关 flink 1.10 sql 方式读取 mysql

ddl 方式 package com.jq.wc import org.apache.flink.streaming.api.scala._

╰+哭是因爲堅強的太久メ/ 2022年12月19日 01:22/ 0 赞/ 85 阅读

相关 flink 1.10 Sql方式读取kafka

ddl模式 import org.apache.flink.streaming.api.scala._ import org.apache.flink.t

Love The Way You Lie/ 2022年11月22日 12:59/ 0 赞/ 154 阅读

相关 flink重复读取mysql_flink读写MySQL的两种方式

2020-05-22 12:09:48,198 WARN org.apache.flink.runtime.webmonitor.handlers.JarPlanHandler

- 日理万妓/ 2022年10月26日 12:58/ 0 赞/ 34 阅读

相关【Flink】Flink SQL 读取 CSV 文件

文章目录 3.7 source 3.7 连续查询 3.8 Sink Tables 3.8.1 案例1

亦凉/ 2022年09月11日 05:18/ 0 赞/ 287 阅读

相关 Flink-04-Flink自定义Source

1、实现并行度为1 的自定义Source 实现SourceFunction接口–一定要指定数据类型 package com.test.streaming.cus

Myth丶恋晨/ 2022年02月25日 06:54/ 0 赞/ 306 阅读

相关 Flink（三）Flink自定义Data Source

一、Flink Kafka source：Flink默认实现的自定义Data Source > 准备工作我们先来看下 Flink 从 Kafka topic 中获取数

柔光的暖阳◎/ 2022年02月19日 03:27/ 0 赞/ 750 阅读

相关 Flink（二）Flink的Data Sources

一、前言 Data Sources 是什么呢？就字面意思其实就可以知道：数据来源。 Flink 做为一款流式计算框架，它可用来做批处理，即处理静态的数据集、历史的数据集

素颜马尾好姑娘i/ 2022年02月19日 03:19/ 0 赞/ 308 阅读

相关【Flink入门】Flink自定义Source读取MySQL数据

在前一篇博客中 [【Flink入门】Flink读取Kafka数据Demo][Flink_Flink_Kafka_Demo] 已经简单介绍了Flink读取Kafka数据并通过De

╰半夏微凉°/ 2021年09月25日 04:44/ 0 赞/ 891 阅读