flink1.10中三种数据处理方式的连接器说明

news2025/1/18 8:43:29

第一种 Streaming（DataStream API）

流式处理的所有的连接器如上图，常用的是kafka、Elasticsearch、Hadoop FileSystem

Kafka连接器依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>

第二种 Batch(DataSet API)

批处理连接器不常用

第三种 Table API&SQL

这是表连接器

JDBC连接器需要的依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-jdbc_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>

直接运行会报错

错误信息中重要提示：
Caused by: java.lang.IllegalArgumentException: JDBC driver class not found.
Caused by: java.lang.ClassNotFoundException: com.mysql.jdbc.Driver

需要添加jdbc驱动

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.44</version>
</dependency>

在这个程序中，流环境使用的是org.apache.flink.streaming.api.environment.StreamExecutionEnvironment，java的流环境

package flink_jdbc_mysql

import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.java.io.jdbc.{JDBCAppendTableSink, JDBCInputFormat}
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.types.Row
import org.apache.flink.streaming.api.datastream.DataStreamSource
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment


object Flink_Mysql_Stream {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val rowTypeInfo: RowTypeInfo = new RowTypeInfo(
      //和mysql中字段的数据一致
      BasicTypeInfo.INT_TYPE_INFO,
      BasicTypeInfo.INT_TYPE_INFO
    )
    val jdbc: JDBCInputFormat = JDBCInputFormat.buildJDBCInputFormat()
      .setDrivername("com.mysql.jdbc.Driver")
      .setDBUrl("jdbc:mysql://localhost/test")
      .setUsername("root")
      .setPassword("root")
      //SELECT COUNT(t.*) from accept_materials m, accept_materials_attachment t WHERE m.materials_id=t.accept_materials_id and m.take_type=2 AND apply_no='';
      .setQuery("select * from e where id = 1")//
      .setRowTypeInfo(rowTypeInfo)
      .finish()

      val datastream: DataStreamSource[Row] = env.createInput(jdbc)

    val sink: JDBCAppendTableSink = JDBCAppendTableSink.builder()
      .setDrivername("com.mysql.jdbc.Driver")
      .setDBUrl("jdbc:mysql://localhost/test")
      .setUsername("root")
      .setPassword("root")
      .setParameterTypes(
        BasicTypeInfo.INT_TYPE_INFO,
        BasicTypeInfo.INT_TYPE_INFO
      )
      .setQuery("insert into a values(?,?)")
      .build()
    /*
    *     emitDataStream需要传入的是org.apache.flink.streaming,api.datastream.DataStream
    *     而这个地方我传入的是org.apache.flink.streaming.api.scala.DataStreaming
    *     导致一致出现类型匹配错误
    *
    *     原因是创建流式环境时
    *     val env = StreamExecutionEnvironment.getExecutionEnvironment
    *     导入的是scala流环境
    *     import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    *     解决办法
    *     导入flink.streaming.api 流环境
    *     import org.apache.flink.streaming.api.datastream.DataStream
    *
    *     创建flink流环境之后
    *     val datastream: DataStreamSource[Row] = env.createInput(jdbc)
    *     返回值类型是DataStreamSource
    *
    *
    *
    * */

    /*
    *  自定义数据流
    * */

    sink.emitDataStream(datastream)

    env.execute()
  }
}

JDBC连接器导入的依赖包，实际上是java语言开发的，使用

env.createInput(jdbc)，这个jdbc是JDBCInputFormat类型的

createInput需要传入的类型是 org.apache.flink.api.common.io.InputForamt

而JDBCInputFormat是InputFormat的子类，所以即使我们使用scala的流式环境，仍然可以使用java的输入流

JDBCInputFormat extends RichInputFormat implements InputFormat

但是使用JDBC连接器以流的方式输出的时候，JDBCAppendOutput类中emitDataStream方法

而方法需要的参数类型DataStream类型是 org.apache.flink.streaming.api.datastream.DataStream，java的流环境

如果我们使用scala创建流环境，使用jdbc连接器以流的方式读取mysql中数据，然后以流的方式输出到mysql中时，则会因为参数类型不匹配导致编译不成功

解决办法就是使用java创建流环境。