Flink之末自定义udf与Sink定义

你的名字 2023-03-13 13:24 53阅读 0赞

# 一、实现UDF函数——更细粒度的控制流 #

## 1.1 函数类（Function Classes） ##

Flink暴露了所有udf函数的接口(实现方式为接口或者抽象类)。例如MapFunction, FilterFunction, ProcessFunction等等。  
下面例子实现了FilterFunction接口：

class FilterFilter extends FilterFunction[String] {
          override def filter(value: String): Boolean = {
    	      value.contains("flink")
          }
    }
    val flinkTweets = tweets.filter(new FlinkFilter)

还可以将函数实现成匿名类

val flinkTweets = tweets.filter(
    	new RichFilterFunction[String] {
    		override def filter(value: String): Boolean = {
    			value.contains("flink")
    		}
    	}
    )

我们filter的字符串"flink"还可以当作参数传进去。

val tweets: DataStream[String] = ...
    val flinkTweets = tweets.filter(new KeywordFilter("flink"))
    class KeywordFilter(keyWord: String) extends FilterFunction[String] {
    	override def filter(value: String): Boolean = {
    		value.contains(keyWord)
    	}
    }

## 1.2 匿名函数（Lambda Functions） ##

val tweets: DataStream[String] = ...
    val flinkTweets = tweets.filter(_.contains("flink"))

## 1.3 富函数（Rich Functions） ##

“富函数”是DataStream API提供的一个函数类的接口，所有Flink函数类都有其Rich版本。它与常规函数的不同在于，可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。

*  RichMapFunction
 *  RichFlatMapFunction
 *  RichFilterFunction
 *  …

Rich Function有一个生命周期的概念。典型的生命周期方法有：

*  `open()`方法是rich function的初始化方法，当一个算子例如map或者filter被调用之前open()会被调用。
 *  `close()`方法是生命周期中的最后一个调用的方法，做一些清理工作。
 *  `getRuntimeContext()`方法提供了函数的RuntimeContext的一些信息，例如函数执行的并行度，任务的名字，以及state状态

class MyFlatMap extends RichFlatMapFunction[Int, (Int, Int)] {
    	var subTaskIndex = 0
    
    	override def open(configuration: Configuration): Unit = {
    		subTaskIndex = getRuntimeContext.getIndexOfThisSubtask
    		// 以下可以做一些初始化工作，例如建立一个和HDFS的连接
    	}
    
    	override def flatMap(in: Int, out: Collector[(Int, Int)]): Unit = {
    		if (in % 2 == subTaskIndex) {
    			out.collect((subTaskIndex, in))
    		}
    	}
    
    	override def close(): Unit = {
    		// 以下做一些清理工作，例如断开和HDFS的连接。
    	}
    }

# 二、Sink定义 #

Flink没有类似于spark中foreach方法，让用户进行迭代的操作。虽有对外的输出操作都要利用Sink完成。最后通过类似如下方式完成整个任务最终输出操作。

stream.addSink(new MySink(xxxx))

官方提供了一部分的框架的sink。除此以外，需要用户自定义实现sink。

## 2.1 Kafka ##

pom.xml文件定义

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

主函数中添加sink：

val union = high.union(low).map(_.temperature.toString)
    union.addSink(new FlinkKafkaProducer011[String]("localhost:9092", "test", new SimpleStringSchema()))

## 2.2 Redis ##

pom.xml文件定义

<dependency>
        <groupId>org.apache.bahir</groupId>
        <artifactId>flink-connector-redis_2.11</artifactId>
        <version>1.0</version>
    </dependency>

定义一个redis的mapper类，用于定义保存到redis时调用的命令：

class MyRedisMapper extends RedisMapper[SensorReading]{
      override def getCommandDescription: RedisCommandDescription = {
        new RedisCommandDescription(RedisCommand.HSET, "sensor_temperature")
      }
      override def getValueFromData(t: SensorReading): String = t.temperature.toString
    
      override def getKeyFromData(t: SensorReading): String = t.id
    }

在主函数中调用：

val conf = new FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build()
    dataStream.addSink( new RedisSink[SensorReading](conf, new MyRedisMapper) )

## 2.3 Elasticsearch ##

pom.xml

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-elasticsearch6_2.11</artifactId>
        <version>1.10.0</version>
    </dependency>

在主函数中调用：

val httpHosts = new util.ArrayList[HttpHost]()
    httpHosts.add(new HttpHost("localhost", 9200))
    
    val esSinkBuilder = new ElasticsearchSink.Builder[SensorReading]( httpHosts, new ElasticsearchSinkFunction[SensorReading] {
      override def process(t: SensorReading, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = {
        println("saving data: " + t)
        val json = new util.HashMap[String, String]()
        json.put("data", t.toString)
        val indexRequest = Requests.indexRequest().index("sensor").`type`("readingData").source(json)
        requestIndexer.add(indexRequest)
        println("saved successfully")
      }
    } )
    dataStream.addSink( esSinkBuilder.build() )

5.6.4 JDBC 自定义sink

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

添加MyJdbcSink

class MyJdbcSink() extends RichSinkFunction[SensorReading]{
      var conn: Connection = _
      var insertStmt: PreparedStatement = _
      var updateStmt: PreparedStatement = _
    
      // open 主要是创建连接
      override def open(parameters: Configuration): Unit = {
        super.open(parameters)
    
        conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "123456")
        insertStmt = conn.prepareStatement("INSERT INTO temperatures (sensor, temp) VALUES (?, ?)")
        updateStmt = conn.prepareStatement("UPDATE temperatures SET temp = ? WHERE sensor = ?")
      }
      // 调用连接，执行sql
      override def invoke(value: SensorReading, context: SinkFunction.Context[_]): Unit = {
        
    	updateStmt.setDouble(1, value.temperature)
        updateStmt.setString(2, value.id)
        updateStmt.execute()
    
        if (updateStmt.getUpdateCount == 0) {
          insertStmt.setString(1, value.id)
          insertStmt.setDouble(2, value.temperature)
          insertStmt.execute()
        }
      }
    
      override def close(): Unit = {
        insertStmt.close()
        updateStmt.close()
        conn.close()
      }
    }

在main方法中增加，把明细保存到mysql中

dataStream.addSink(new MyJdbcSink())