[DB] Flink 读 MySQL

小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

思路

在 Flink 中创建一张表有两种方法：

从一个文件中导入表结构（Structure）（常用于批计算）（静态）
从 DataStream 或者 DataSet 转换成 Table （动态）

package com.kaikeba.mysql.demo

import org.apache.flink.api.common.typeinfo.BasicTypeInfo

import org.apache.flink.api.java.io.jdbc.JDBCInputFormat

import org.apache.flink.api.java.typeutils.RowTypeInfo

import org.apache.flink.api.scala._

import org.apache.flink.table.api.scala.BatchTableEnvironment

import org.apache.flink.types.Row

object Flink2Mysql {

  def main(args: Array[String]): Unit = {

    //设定执行环境

    val env = ExecutionEnvironment.getExecutionEnvironment

    val tEnv = BatchTableEnvironment.create(env)

    //通过创建JDBCInputFormat读取JDBC数据源

    val jdbcDataSet: DataSet[Row] =

      env.createInput(JDBCInputFormat.buildJDBCInputFormat()

        .setDrivername("com.mysql.cj.jdbc.Driver")

        .setDBUrl("jdbc:mysql://127.0.0.1:3306/flink-mysql?serverTimezone=GMT%2B8&characterEncoding=UTF-8&useSSL=false")

        .setUsername("root")

        .setPassword("Chen1227+")

        .setQuery("select * from filter")

        .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO))

        .finish()

      )

    //将DataSet注册为表

    tEnv.registerDataSet("tb", jdbcDataSet)

    //执行查询操作

    val table = tEnv.sqlQuery("select * from tb")

    //把table转为DataSet

    tEnv.toDataSet[Row](table).print()

  }

}