Flink DataSouce

flink流处理和批处理都内置了很多数据源，可以满足我们大部分使用场景，当然也可以通过实现flink提供的接口来实现其他数据源的接入。接下来我们就分别来了解下flink批处理和流处理的数据源吧。

1 批处理

批处理的数据源主要大致分为两类：集合数据源、文件数据源。

sequence

集合数据源之迭代类。平时我们所使用到的scala中Seq(),Array(),List(),Map(),Set()等都是迭代类。

    env.fromCollection(data: Iterable[T])
    // 示例
    env.fromCollection(Seq(1, 2, 3))
    env.fromCollection(Array(1, 2, 3, 4, 5))
    env.fromCollection(List(1, 2, 3, 4, 5))
    env.fromCollection(Map("a" -> 1, "b" -> 2, "c" -> 3))
    env.fromCollection(Set(3, 5, 6))

    // 并行版本
    env.fromParallelCollection(iterator: SplittableIterator[T])
    // 示例
    env.fromParallelCollection(new NumberSequenceIterator(3, 19))

01_parallel_collection.png

集合类型源之元素数组。类似于java中的可变参数列表。

    env.fromElements(data: T*)
    // 示例
    env.fromElements(1, 2, 3)
    env.fromElements("a", "b", "c")

file

一般来说，我们的批处理程序的数据源主要都是文件数据源，包括本地文件系统中的和分布式文件系统中的文件。使用频次比较高的是readTextFile()和readCsvFile()，当然也可以通过重写readFlie(new FileInputStream[T], filePath: String)中的FileInputFormat来读取不同格式文件。

    env.readFile(new FileInputStream[T], filePath: String)
    env.readTextFile(filePath: String, charsetName: String = "UTF-8")
    env.readTextFileWithValue(filePath: String, charsetName: String = "UTF-8")
    env.readFileOfPrimitives(filePath: String, delimiter: String = "\n")
    env.readCsvFile(
      filePath: String,                 // csv文件地址
      lineDelimiter: String = "\n",     // 行划分符号
      fieldDelimiter: String = ",",     // 字段划分符号
      quoteCharacter: Character = null,
      ignoreFirstLine: Boolean = false, // 忽略首行
      ignoreComments: String = null,
      lenient: Boolean = false,         // 是否对数据严格判断。false表示严格判断，缺失的数据则会被忽略
      includedFields: Array[Int] = null,
      pojoFields: Array[String] = null
    )

readFile，自定义读取文件内容的逻辑

    env.readFile(new MyFileInputFormat[String](), getPath("words.txt"))

// 重写读取文件的逻辑
class MyFileInputFormat extends FileInputFormat[String]() {
  private var end = false
  private var input: BufferedReader = _

  // 只执行一次，此处创建输入流对象
  override def open(fileSplit: FileInputSplit): Unit = {
    val inputStream = new FileInputStream(this.getFilePath.getPath)
    input = new BufferedReader(new InputStreamReader(inputStream))
  }

  // 判断是否读取到了文件尾部
  override def reachedEnd(): Boolean = this.end

  // 读取下一行的操作逻辑
  override def nextRecord(ot: String): String = {
    val str: String = input.readLine()
    if (str == null) {
      this.end = true
      ""
    } else str
  }
}

2 流处理

flink流处理的数据源大致分为四种：集合数据源、文件数据源、套接字数据源、自定义数据源。flink还原生给我们写好了kafka数据源，这是实时流处理中使用的最频繁没有之一的数据源(下边会单独简单展示一下)。

sequence

不用多说了吧，完全参考批处理中的sequence。

    env.fromCollection(data: Iterable[T])
    env.fromElements(data: T*)
    env.fromParallelCollection(iterator: SplittableIterator[T])

file

一样参考批处理中的flie。只是有一点儿不同的是流处理中readFile()还能监视文件的变更状况来进行更多的处理方式，如文件新增了记录，可以重新处理文件或者直接退出。

    env.readTextFile(filePath: String, charsetName: String)
    env.readFile(
      inputFormat: FileInputFormat[T],
      filePath: String,
      watchType: FileProcessingMode,    // 监视路径并响应新数据，或处理一次并退出
      interval: Long)                   // 间隔时间millis

socket

直接监视指定机子socket端口的记录。

    env.socketTextStream(
      hostname: String, 
      port: Int, 
      delimiter: Char = '\n', 
      maxRetry: Long = 0)               // 如果端口监听中断，最大重试间隔时间

自定义

当上边所有的数据源都满足不了我们的场景需求时，我们可以通过继承flink暴露的SourceFunction来实现自己的数据源(下文会展示自定义MySQL数据源)。

    env.addSource(function: SourceContext[T] => Unit)
    env.addSource(function: SourceFunction[T])

3 kafka connector

kafka connector是flink提供给我们的自定义连接器，可以直接实例化FlinkKafkaConsumer对象来消费kafka中记录。

object FlinkStreamDataSource {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // kafka properties参数
    val props = new Properties()
    props.setProperty("bootstrap.servers", "172.16.57.101:9092,172.16.57.102:9092,172.16.57.103:9092")
    props.setProperty("zookeeper.connect", "172.16.57.101:2181,172.16.57.102:2181,172.16.57.103:2181")
    props.setProperty("group.id", "leslie")
    props.setProperty("auto.offset.reset", "latest")

    val original: DataStream[String] = env
      .addSource(new FlinkKafkaConsumer010[String]("test", new SimpleStringSchema(), props))

    original
      .flatMap(_.split(","))
      .map(_ + "_test")
      .print()

    env.execute("flink_streaming_data_source")
  }
}

01_datasource_kafka.png

4 自定义DataSource

想要实现自定义的数据源十分简单，只需继承flink的SourceFunction接口并重写其中的run(),cancel()方法。

object FlinkStreamCustomerDataSource {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 自定义dataSource
    env.addSource(new MysqlDataSource).print()

    env.execute("customer_date_source")
  }
}

下文代码MysqlDataSource类是继承了RichSourceFunction类，RichSourceFunction实现SourceFunction接口并同时继承AbstractRichFunction抽象类，AbstractRichFunction抽象类又实现RichFunction接口。为什么我们的自定义数据源要多继承RichSourceFunction类呢？原因就在这个Rich上！我们将这样带有"Rich"前缀的函数类称为富函数，既然是富函数了那么一定是比普通的函数多给我们带来一些功能。RichSourceFunction中的open(),close()就是“富”出来的方法，open()方法仅在函数类实例化的时候调用一次(通常用来建立连接)，close()则是在实例对象销毁前调用一次(通常用来关闭连接)，可以避免重复进行创建连接销毁连接操作。(当然富函数不仅仅只“富”了这么一点点，还“富”出来运行时上下文，这可是个好东西。此处不扩展哦，以后用到再来讨论)

02_rich_function.png

我们需要进行获取外部存储组件数据的操作就在SourceFunction的run(),cancel()方法中实现。

class MysqlDataSource extends RichSourceFunction[String] {
  private var pStmt: PreparedStatement = _
  private var conn: Connection = _

  // 开始方法，只执行一次，建立和mysql的连接
  override def open(parameters: Configuration): Unit = {
    Class.forName("com.mysql.jdbc.Driver")
    val url = "jdbc:mysql://localhost:3306/test_for_mysql?useSSL=false"
    val username = "root"
    val password = "123456"
    conn = DriverManager.getConnection(url, username, password);
    val sql =
      """
        |select id, name, sex, age from user;
        |""".stripMargin
    pStmt = conn.prepareStatement(sql)
  }

  // 结束方法，只执行一次，关闭连接
  override def close(): Unit = {
    // 关闭连接
    if (pStmt != null) pStmt.close()
    if (conn != null) conn.close()
  }

  // 主体执行方法
  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
    val rs: ResultSet = pStmt.executeQuery()
    while (rs.next()) {
      val id: Int = rs.getInt("id")
      val name: String = rs.getString("name")
      val sex: String = rs.getString("sex")
      val age: Int = rs.getInt("age")

      ctx.collect(s"id: $id, name: $name, sex: $sex, age:$age") // 收集记录到上下文中
    }
  }
  override def cancel(): Unit = {}
}

02_customer_mysql.png