大数据之Flink(5) | Sink

约定不等于承诺〃 2022-12-16 09:10 223阅读 0赞

### Sink ###

*  Kafka
 *  Redis
 *  Elasticsearch
 *  JDBC 自定义sink
 *  File
 *  教你如何看官网源码

# Kafka #

依赖

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

测试数据

sensor_1,1547718199,35.8
    sensor_6,1547718201,15.4
    sensor_7,1547718202,6.7
    sensor_10,1547718205,38.1
    sensor_1,1547718207,37.2
    sensor_1,1547718212,33.5
    sensor_1,1547718215,38.1

**从文件中读取文件测试**

import learning.SensorReading
    import org.apache.flink.api.common.serialization.SimpleStringSchema
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.connectors.kafka. FlinkKafkaProducer011
    
    //输入数据的样例类
    case class SensorReading(id:String,timestamp:Long,temperature:Double)
    
    object KafkaSinkTest { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        val inputStream = env.readTextFile("D:\\idea\\Flink-Learning\\src\\main\\resources\\sensor.txt")
        val dataStream = inputStream
          .map(data => { 
            val dataArray = data.split(",")
            SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble).toString
          })
        
        dataStream.addSink(new FlinkKafkaProducer011[String]("hadoop12:9092", "sinkTest", new SimpleStringSchema()))
    
        env.execute("sink test")
      }
    }

在hadoop12输入，并运行idea

bin/kafka-console-consumer.sh --bootstrap-server hadoop12:9092 --topic sinkTest

查看结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]  
kafka生产者->消费者测试

import java.util.Properties
    import learning.SensorReading
    import org.apache.flink.api.common.serialization.SimpleStringSchema
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.connectors.kafka.{ FlinkKafkaConsumer011, FlinkKafkaProducer011}
    
    //输入数据的样例类
    case class SensorReading(id:String,timestamp:Long,temperature:Double)
    
    object KafkaSInkTest2 { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        val properties = new Properties()
        properties.setProperty("bootstrap.servers", "hadoop12:9092")
        properties.setProperty("group.id", "consumer-group")
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
        properties.setProperty("auto.offset.reset", "latest")
        val inputStream = env.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))
    
        val dataStream: DataStream[String] = inputStream
          .map(data => { 
            val dataArray = data.split(",")
            SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble).toString
          })
        inputStream.print()
        dataStream.addSink(new FlinkKafkaProducer011[String]("hadoop12:9092","sinkTest",new SimpleStringSchema()))
    
        env.execute("sink test")
      }
    
    }

在hadoop12输入，并运行idea

bin/kafka-console-producer.sh --broker-list hadoop12:9092 --topic sensor
     bin/kafka-console-consumer.sh --bootstrap-server hadoop12:9092 --topic sinkTest

测试结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]

# Redis #

import learning.SensorReading
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.connectors.redis.RedisSink
    import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
    import org.apache.flink.streaming.connectors.redis.common.mapper.{ RedisCommand, RedisCommandDescription, RedisMapper}
    
    object RedisSinkTest { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        val inputStream = env.readTextFile("D:\\idea\\Flink-Learning\\src\\main\\resources\\sensor.txt")
        val dataStream = inputStream
          .map(data => { 
            val dataArray = data.split(",")
            SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
          })
    
        //定义一个redis的配置类
        val conf = new FlinkJedisPoolConfig.Builder()
          .setHost("hadoop12")
          .setPort(6379)
          .build()
        //定义一个RedisMapper
        val myMapper = new RedisMapper[SensorReading] { 
          //定义保存数据到redis的命令，hset table_name key value
          override def getCommandDescription: RedisCommandDescription = { 
            new RedisCommandDescription(RedisCommand.HSET,"sensor_temp")
          }
    
          override def getKeyFromData(data: SensorReading): String = data.id
    
          override def getValueFromData(data: SensorReading): String = data.temperature.toString
        }
        dataStream.addSink(new RedisSink[SensorReading](conf,myMapper))
    
        env.execute("redis sink test")
      }
    }

启动redis

redis-server redis.conf
    redis-cli

查看结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]  
**思考数据有7条数据，为什么只有4条数据？**  
因为hash表，显示时间戳最新的数据

# Elasticsearch #

import java.util
    
    import learning.SensorReading
    import org.apache.flink.api.common.functions.RuntimeContext
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.connectors.elasticsearch.{ ElasticsearchSinkFunction, RequestIndexer}
    import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
    import org.apache.http.HttpHost
    import org.elasticsearch.client.Requests
    
    object esSinkTest { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        val inputStream = env.readTextFile("D:\\idea\\Flink-Learning\\src\\main\\resources\\sensor.txt")
        val dataStream = inputStream
          .map(data => { 
            val dataArray = data.split(",")
            SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
          })
    
        val httpHosts = new util.ArrayList[HttpHost]()
        httpHosts.add(new HttpHost("hadoop12",9200))
        //定义一个ElasticsearchSinkFunction
        val esSinkFunc = new ElasticsearchSinkFunction[SensorReading] { 
          override def process(t: SensorReading, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = { 
            //包装写入es的数据
            val dataSource = new util.HashMap[String, String]()
            dataSource.put("sensor_id",t.id)
            dataSource.put("temp",t.temperature.toString)
            dataSource.put("ts",t.timestamp.toString)
            //创建一个index request
            val indexRequest = Requests
              .indexRequest()
              .index("sensor_temp")
              .`type`("readingdata")
              .source(dataSource)
            //用indexer发送http请求
            requestIndexer.add(indexRequest)
            println(t + "saved successfully")
    
          }
        }
        dataStream.addSink(new ElasticsearchSink.Builder[SensorReading](httpHosts,esSinkFunc).build())
    
        env.execute("es sink test")
    
    
      }
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]

# JDBC 自定义sink #

依赖

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

import java.sql.{ Connection, DriverManager, PreparedStatement}
    import learning.SensorReading
    import org.apache.flink.configuration.Configuration
    import org.apache.flink.streaming.api.functions.sink.{ RichSinkFunction, SinkFunction}
    import org.apache.flink.streaming.api.scala._
    
    object JdbcSinkTest { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        val inputStream = env.readTextFile("D:\\idea\\Flink-Learning\\src\\main\\resources\\sensor.txt")
        val dataStream = inputStream
          .map(data => { 
            val dataArray = data.split(",")
            SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
          })
    
        dataStream.addSink(new MyJdbcSink())
    
        env.execute("jdbc sink test")
    
    
      }
    
    }
    //自定义一个SinkFunction
    class MyJdbcSink() extends RichSinkFunction[SensorReading]{ 
      //首先定义sql连接，以及预编译语句
      var conn : Connection = _
      var insertStmt:PreparedStatement = _
      var updateStmt:PreparedStatement = _
    
      //在open生命周期方法中创建连续以及预编译语句
      override def open(parameters: Configuration): Unit = { 
        conn = DriverManager.getConnection("jdbc:mysql://hadoop12:3306/test","root","yy8266603")
        insertStmt = conn.prepareStatement("insert into temp (sensor, temperature) values (?,?)")
        updateStmt = conn.prepareStatement("update temp set temperature = ? where sensor = ?")
      }
    
      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()
      }
    }

结果  
显示时间戳最大的数据  
![在这里插入图片描述][20201019214118845.png_pic_center]

# File #

import learning.SensorReading
    import org.apache.flink.api.common.serialization.SimpleStringEncoder
    import org.apache.flink.core.fs.Path
    import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
    import org.apache.flink.streaming.api.scala._
    
    object FileSinkTest { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        val inputStream = env.readTextFile("D:\\idea\\Flink-Learning\\src\\main\\resources\\sensor.txt")
        val dataStream = inputStream
          .map(data => { 
            val daraArray = data.split(",")
            SensorReading(daraArray(0), daraArray(1).toLong, daraArray(2).toDouble).toString
          })
    
        dataStream.addSink(
          StreamingFileSink.forRowFormat(
            new Path("D:\\idea\\Flink-Learning\\src\\main\\resources\\out.txt"),
            new SimpleStringEncoder[String]("UTF-8"))
        .build())
    
        env.execute("sink test")
      }
    
    }

结果  
![在这里插入图片描述][20201019175754735.png_pic_center]

# 教你如何看官网源码 #

1.首先打开官网  
[https://flink.apache.org/flink-architecture.html][https_flink.apache.org_flink-architecture.html]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 4]  
2.自己什么版本，就改成与自己对应的版本！(这里我就以1.11为列)  
![在这里插入图片描述][20201019180023934.png_pic_center]  
3.这里应有尽有，但是我们这里需要看的是sink  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 6]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/04b75300558044e19b2b55284442bb45.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/180017a071d04c31a0558ed2af48132f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/64b7cb7e71e546c29b10065f97c0c196.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/742e76ab219d486d8af469805e45d0b8.png
[20201019214118845.png_pic_center]: /images/20221123/c73ffb7592c14e1399ca67f5d00f4cb1.png
[20201019175754735.png_pic_center]: /images/20221123/675635c6dcd6433187e8e5d9e56a96c1.png
[https_flink.apache.org_flink-architecture.html]: https://flink.apache.org/flink-architecture.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/623b06d7c583477d8fc2e4f1dbab7bbf.png
[20201019180023934.png_pic_center]: /images/20221123/b205823f489847559c1b06b0a967316a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/ef899e0bbc434099ab1e1c2232358aa1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/e7b8399bc7124a32a154ced4e5d39052.png