Flink章节之对接mysql数据源

发布于:2021-12-03 02:43:56

主要针对 flink 1.11 scala 版本,各版本可能存在细节差异。


直接上代码


package net.cn.ablue.flink.scala.read

import java.sql.{Connection, DriverManager, PreparedStatement}

import net.cn.ablue.scala.jdbc.conf.{JdbcConnectionInfo, JdbcHelper}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.source.RichSourceFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

/**
* @author Ablue
* @version 1.0.0
* @ClassName FlinkSourceMysql.scala
* @projectName operator-flink-mysql
* @createTime 2021年01月26日 22:55:00
* @see `flin对接 mysql`
*/
object FlinkSourceMysql {


def main(args: Array[String]): Unit = {
// 创建一些 flink 初始的参数
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
// 指定程序的并行度
env.setParallelism(1)
// 这里可以做一些其他业务系统的对接整合
// 自定义mysqlSource
val source: DataStream[(Int, String, String)] = env.addSource(new FlinkMysqlSource)


source.print()
env.execute()
}
}
/**
* @author Ablue
* @version 1.0.0
* @ClassName FlinkMysqlSource.scala
* @projectName operator-flink-mysql
* @createTime 2021年01月26日 22:55:00
* @see `flin对接 mysql`
*/
class FlinkMysqlSource extends RichSourceFunction[(Int, String, String)] {
/**
* jdbc 连接信息对象获取
*/
val JdbcConnInfo: JdbcConnectionInfo = JdbcHelper.getJdbcConnectionInfo()

/**
*
* @param ctx
*/
override def run(ctx: SourceContext[(Int, String, String)]): Unit = {
/**
* 这里其实就是jdbc 编程
*/
val connection: Connection =
DriverManager.getConnection(JdbcConnInfo.url, JdbcConnInfo.properties)

val sql = "select uid,uname,time from user"
val ps: PreparedStatement = connection.prepareStatement(sql)
val queryRequest = ps.executeQuery()
while (queryRequest.next()) {
// 获取字段内容
val id = queryRequest.getInt("uid")
val username = queryRequest.getString("uname")
val password = queryRequest.getString("time")
// 返回的 sourceContxt 内容,即 mysql 表中字段内容
ctx.collect((id, username, password))
}

}

/**
* 这里可以对开启的数据库连接做关闭
*/
override def cancel(): Unit = {}
}


flink 对接 mysql 的话,主要是需要对 RichSourceFunction 进行自定义数据源的实现.


RichSourceFunction 要重写的方法及其作用


run:对接数据源时需要做的一些操作,例如:需要对接 mysql,这里是通过jdbc的方式进行连接、


处理、最终、返回结果;结果就是 SourceContext[(Int, String, String)]) 数据是以元组进行返回。


cancel:run运行之后的方法,可以在这里进行资源释放


一些工具类提供参考


package net.cn.ablue.scala.jdbc.conf

import java.util.Properties

import com.typesafe.config.{Config, ConfigFactory}
import net.cn.ablue.java.util.JdbcConstants

/**
* jdbc 管理
* @Author Ablue
*/
object JdbcHelper{
/**
* 获取 jdbc 连接信息
* @return
*/
def getJdbcConnectionInfo():JdbcConnectionInfo ={
// 加载一些配置信息
val conf: Config = ConfigFactory.load()

Class.forName(conf.getString(JdbcConstants.ConnectionProperties.DRIVER))

val url = conf.getString(JdbcConstants.ConnectionProperties.URL)
val user = conf.getString(JdbcConstants.ConnectionProperties.USER)
val pwd = conf.getString(JdbcConstants.ConnectionProperties.PWD)

val properties = new Properties()

// 第一个参数是链接数据库的参数不能随意写
properties.setProperty("user", user)
properties.setProperty("password", pwd)
// 将配置信息的样例类返回
JdbcConnectionInfo(url, user, pwd, properties)
}
}


/**
* @author Ablue
* @version 1.0.0
* @ClassName JdbcConnectionInfo.scala
* @projectName operator-flink-mysql
* @createTime 2021年01月26日 22:55:00
* @see `mysql 连接信息样例类`
*/
case class JdbcConnectionInfo(url: String, userName: String, password: String, properties: Properties)

package net.cn.ablue.java.util;

/**
* @author Ablue
* @version 1.0.0
* @ClassName JdbcConstants.java
* @projectName operator-flink-mysql
* @createTime 2021年01月26日 22:55:00
* @see `操作jdbc的常量类`
*/
public interface JdbcConstants {
// 这里其实放一些连接池的信息
class Pools{

}

/**
* 连接信息的 properties 文件
*/
class ConnectionProperties{
public static final String URL = "jdbc.url";
public static final String USER = "jdbc.user";
public static final String PWD = "jdbc.pwd";
public static final String DRIVER = "jdbc.driver";
// public static final String ORACLE_DRIVER = "jdbc.mysql.driver";
}

}

application.conf


jdbc.url="jdbc:mysql://localhost:3306/study?characterEncoding=utf-8"
jdbc.user="root"
jdbc.pwd="root"
jdbc.driver="com.mysql.jdbc.Driver"

maven pom.xml 提供



xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
4.0.0

cn.net.ablue
operator-flink-mysql
1.0-SNAPSHOT

UTF-8
2.11.8
3.1.4
1.11.0
5.1.42
2.5.0




org.scala-lang
scala-library
${scala.version}



org.apache.flink
flink-streaming-scala_2.11
${flink.version}


org.apache.flink
flink-scala_2.11
${flink.version}


org.apache.flink
flink-clients_2.11
${flink.version}


org.apache.hadoop
hadoop-client
${hadoop.version}


mysql
mysql-connector-java
${jdbc.version}



org.scalikejdbc
scalikejdbc_2.11
${scalikejdbc.version}




org.scalikejdbc
scalikejdbc-config_2.11
${scalikejdbc.version}





-------------------------------------------end----------------------------------------------------

相关推荐

最新更新

猜你喜欢