devbox
小蓝xlanll
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

4.1.15 Flink-流处理框架-Flink流处理API之sink输出操作_streamsource.addsink()

作者：小蓝xlanll | 2024-05-24 01:04:41

赞

踩

streamsource.addsink()

目录

1.写在前面

2.输出到Kafka（source/sink）

3.输出到Redis（sink）

4.输出到Elasticsearch（sink）

5.JDBC 自定义 sink

1.写在前面

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

stream.addSink(new MySink(xxxx))

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

2.输出到Kafka（source/sink）

pom.xml

  1. <!--
  2. https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka-0.11
  3. -->
  4. <dependency>
  5.     <groupId>org.apache.flink</groupId>
  6.     <artifactId>flink-connector-kafka-0.11_2.12</artifactId>
  7.     <version>1.10.1</version>
  8. </dependency

主函数中添加 sink：

dataStream.addSink(new FlinkKafkaProducer011[String]("localhost:9092","test", new SimpleStringSchema()))

 完整代码：

kafka生产者9092--->flink---->kafka消费者

  1. package com.atguigu.apitest.sink;
  2.  
  3. import com.atguigu.apitest.beans.SensorReading;
  4. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
  8. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
  9.  
  10. import java.util.Properties;
  11.  
  12. public class SinkTest1_Kafka {
  13.     public static void main(String[] args) throws Exception{
  14.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  15.         env.setParallelism(1);
  16.  
  17. //        // 从文件读取数据
  18. //        DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
  19.  
  20.         Properties properties = new Properties();
  21.         properties.setProperty("bootstrap.servers", "localhost:9092");
  22.         properties.setProperty("group.id", "consumer-group");
  23.         properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  24.         properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  25.         properties.setProperty("auto.offset.reset", "latest");
  26.  
  27.         // 从文件读取数据
  28.         DataStream<String> inputStream = env.addSource( new FlinkKafkaConsumer011<String>("sensor", new SimpleStringSchema(), properties));
  29.  
  30.         // 转换成SensorReading类型
  31.         DataStream<String> dataStream = inputStream.map(line -> {
  32.             String[] fields = line.split(",");
  33.             return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2])).toString();
  34.         });
  35.  
  36.         dataStream.addSink( new FlinkKafkaProducer011<String>("localhost:9092", "sinktest", new SimpleStringSchema()));
  37.  
  38.         env.execute();
  39.     }
  40. }

启动一个kafka消费者消费flink输出的数据

 

3.输出到Redis（sink）

pom.xml

  1. <!-- https://mvnrepository.com/artifact/org.apache.bahir/flink-connector-redis-->
  2. <dependency>
  3.     <groupId>org.apache.bahir</groupId>
  4.     <artifactId>flink-connector-redis_2.11</artifactId>
  5.     <version>1.0</version>
  6. </dependency>

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

  1. public static class MyRedisMapper implements RedisMapper<SensorReading>{
  2.     // 保存到 redis 的命令，存成哈希表
  3.     public RedisCommandDescription getCommandDescription() {
  4.         return new RedisCommandDescription(RedisCommand.HSET, "sensor_tempe");
  5.     }
  6.     public String getKeyFromData(SensorReading data) {
  7.         return data.getId();
  8.     }
  9.     public String getValueFromData(SensorReading data) {
  10.         return data.getTemperature().toString();
  11.     }
  12. }

在主函数中调用：

  1. FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build();
  2.  
  3. dataStream.addSink( new RedisSink<SensorReading>(config, new MyRedisMapper()) );

完整代码：

  1. package com.atguigu.apitest.sink;
  2.  
  3. import com.atguigu.apitest.beans.SensorReading;
  4. import org.apache.flink.streaming.api.datastream.DataStream;
  5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  6. import org.apache.flink.streaming.connectors.redis.RedisSink;
  7. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
  8. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
  9. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
  10. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
  11.  
  12. public class SinkTest2_Redis {
  13.     public static void main(String[] args) throws Exception {
  14.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  15.         env.setParallelism(1);
  16.  
  17.         // 从文件读取数据
  18.         DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
  19.  
  20.         // 转换成SensorReading类型
  21.         DataStream<SensorReading> dataStream = inputStream.map(line -> {
  22.             String[] fields = line.split(",");
  23.             return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
  24.         });
  25.  
  26.         // 定义jedis连接配置
  27.         FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder()
  28.                 .setHost("localhost")
  29.                 .setPort(6379)
  30.                 .build();
  31.  
  32.         dataStream.addSink( new RedisSink<>(config, new MyRedisMapper()));
  33.  
  34.         env.execute();
  35.     }
  36.  
  37.     // 自定义RedisMapper
  38.     public static class MyRedisMapper implements RedisMapper<SensorReading>{
  39.         // 定义保存数据到redis的命令，存成Hash表，hset sensor_temp id temperature
  40.         @Override
  41.         public RedisCommandDescription getCommandDescription() {
  42.             return new RedisCommandDescription(RedisCommand.HSET, "sensor_temp");
  43.         }
  44.  
  45.         @Override
  46.         public String getKeyFromData(SensorReading data) {
  47.             return data.getId();
  48.         }
  49.  
  50.         @Override
  51.         public String getValueFromData(SensorReading data) {
  52.             return data.getTemperature().toString();
  53.         }
  54.     }
  55. }

 

4.输出到Elasticsearch（sink）

pom.xml

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-connector-elasticsearch6_2.12</artifactId>
  4.     <version>1.10.1</version>
  5. </dependency>

在主函数中调用：

  1. // es 的 httpHosts 配置
  2. ArrayList<HttpHost> httpHosts = new ArrayList<>();
  3. httpHosts.add(new HttpHost("localhost", 9200));
  4. dataStream.addSink( new ElasticsearchSink.Builder<SensorReading>(httpHosts, new MyEsSinkFunction()).build());

ElasitcsearchSinkFunction 的实现：

  1. public static class MyEsSinkFunction implements ElasticsearchSinkFunction<SensorReading>{
  2.     @Override
  3.     public void process(SensorReading element, RuntimeContext ctx, RequestIndexer indexer){
  4.  
  5.         HashMap<String, String> dataSource = new HashMap<>();
  6.         dataSource.put("id", element.getId());
  7.         dataSource.put("ts", element.getTimestamp().toString());
  8.         dataSource.put("temp", element.getTemperature().toString());
  9.         IndexRequest indexRequest = Requests.indexRequest().index("sensor").type("readingData").source(dataSource);
  10.         indexer.add(indexRequest);
  11.     }
  12. }

完整代码：

  1. package com.atguigu.apitest.sink;
  2.  
  3. import com.atguigu.apitest.beans.SensorReading;
  4. import org.apache.flink.api.common.functions.RuntimeContext;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
  8. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
  9. import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
  10. import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
  11. import org.apache.http.HttpHost;
  12. import org.elasticsearch.action.index.IndexRequest;
  13. import org.elasticsearch.client.Requests;
  14.  
  15. import java.util.ArrayList;
  16. import java.util.HashMap;
  17.  
  18. public class SinkTest3_Es {
  19.     public static void main(String[] args) throws Exception {
  20.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  21.         env.setParallelism(1);
  22.  
  23.         // 从文件读取数据
  24.         DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
  25.  
  26.         // 转换成SensorReading类型
  27.         DataStream<SensorReading> dataStream = inputStream.map(line -> {
  28.             String[] fields = line.split(",");
  29.             return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
  30.         });
  31.  
  32.         // 定义es的连接配置
  33.         ArrayList<HttpHost> httpHosts = new ArrayList<>();
  34.         httpHosts.add(new HttpHost("localhost", 9200));
  35.  
  36.         dataStream.addSink(new ElasticsearchSink.Builder<SensorReading>(httpHosts, new MyEsSinkFunction()).build());
  37.  
  38.         env.execute();
  39.     }
  40.  
  41.     // 实现自定义的ES写入操作
  42.     public static class MyEsSinkFunction implements ElasticsearchSinkFunction<SensorReading>{
  43.         @Override
  44.         public void process(SensorReading element, RuntimeContext ctx, RequestIndexer indexer) {
  45.             // 定义写入的数据source
  46.             HashMap<String, String> dataSource = new HashMap<>();
  47.             dataSource.put("id", element.getId());
  48.             dataSource.put("temp", element.getTemperature().toString());
  49.             dataSource.put("ts", element.getTimestamp().toString());
  50.  
  51.             // 创建请求，作为向es发起的写入命令
  52.             IndexRequest indexRequest = Requests.indexRequest()
  53.                     .index("sensor")
  54.                     .type("readingdata")
  55.                     .source(dataSource);
  56.  
  57.             // 用index发送请求
  58.             indexer.add(indexRequest);
  59.         }
  60.     }
  61. }

5.JDBC 自定义 sink（写入到mysql）

  1. <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
  2. <dependency>
  3.     <groupId>mysql</groupId>
  4.     <artifactId>mysql-connector-java</artifactId>
  5.     <version>5.1.44</version>
  6. </dependency>

添加 MyJdbcSink

  1. package com.atguigu.apitest.sink;
  2. import com.atguigu.apitest.beans.SensorReading;
  3. import com.atguigu.apitest.source.SourceTest4_UDF;
  4. import org.apache.flink.configuration.Configuration;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
  9. import org.apache.flink.streaming.api.functions.sink.SinkFunction;
  10.  
  11. import java.sql.Connection;
  12. import java.sql.DriverManager;
  13. import java.sql.PreparedStatement;
  14.  
  15. public class SinkTest4_Jdbc {
  16.     public static void main(String[] args) throws Exception {
  17.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  18.         env.setParallelism(1);
  19.  
  20.         // 从文件读取数据
  21. //        DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
  22. //
  23. //        // 转换成SensorReading类型
  24. //        DataStream<SensorReading> dataStream = inputStream.map(line -> {
  25. //            String[] fields = line.split(",");
  26. //            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
  27. //        });
  28.  
  29.         DataStream<SensorReading> dataStream = env.addSource(new SourceTest4_UDF.MySensorSource());
  30.  
  31.         dataStream.addSink(new MyJdbcSink());
  32.  
  33.         env.execute();
  34.     }
  35.  
  36.     // 实现自定义的SinkFunction
  37.     public static class MyJdbcSink extends RichSinkFunction<SensorReading> {
  38.         // 声明连接和预编译语句
  39.         Connection connection = null;
  40.         PreparedStatement insertStmt = null;
  41.         PreparedStatement updateStmt = null;
  42.  
  43.         @Override
  44.         public void open(Configuration parameters) throws Exception {
  45.             connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "123456");
  46.             insertStmt = connection.prepareStatement("insert into sensor_temp (id, temp) values (?, ?)");
  47.             updateStmt = connection.prepareStatement("update sensor_temp set temp = ? where id = ?");
  48.         }
  49.  
  50.         // 每来一条数据，调用连接，执行sql
  51.         @Override
  52.         public void invoke(SensorReading value, Context context) throws Exception {
  53.             // 直接执行更新语句，如果没有更新那么就插入
  54.             updateStmt.setDouble(1, value.getTemperature());
  55.             updateStmt.setString(2, value.getId());
  56.             updateStmt.execute();
  57.             if( updateStmt.getUpdateCount() == 0 ){
  58.                 insertStmt.setString(1, value.getId());
  59.                 insertStmt.setDouble(2, value.getTemperature());
  60.                 insertStmt.execute();
  61.             }
  62.         }
  63.  
  64.         @Override
  65.         public void close() throws Exception {
  66.             insertStmt.close();
  67.             updateStmt.close();
  68.             connection.close();
  69.         }
  70.     }
  71. }

在 main 方法中增加，把明细保存到 mysql 中，而且写入的都是最新的实时数据。

dataStream.addSink(new MyJdbcSink())

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小蓝xlanll/article/detail/615093
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号