Flink-05 Flink Java 3分钟上手 Redis FlinkJedisPoolConfig 从Kafka写入Re ...

代码堆栈

会同步代码到 GitHub
https://github.com/turbo-duck/flink-demo

内容先容

上节我们已经实现了，对Kafka数据的消费和盘算，最终把结果输出到了控制台上。如下图：
Kafka In Docker

TestKafkaProducer

将数据写入到Kafka中的结果

FlinkConsumer

Flink消费Kafka的结果如下图，已经按照我们的需求进行盘算了。

这节内容

本节依然使用Flink对Kafka进行消费，但与上节差别的是（上节将结果输出到控制台上），本节将把Flink盘算的结果输出到Redis中进行生存（当然也可以存储到别的地方，这里以Redis为例）。
pom.xml

重点关注 flink-connector-redis_2.11 这个包。这是Redis相关的依赖。

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0"
3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5.     <modelVersion>4.0.0</modelVersion>
7.     <groupId>org.example</groupId>
8.     <artifactId>flink-demo-01</artifactId>
9.     <version>1.0-SNAPSHOT</version>
11.     <properties>
12.         <maven.compiler.source>8</maven.compiler.source>
13.         <maven.compiler.target>8</maven.compiler.target>
14.         <flink.version>1.13.2</flink.version>
15.         <scala.binary.version>2.12</scala.binary.version>
16.     </properties>
18.     <dependencies>
19.         <dependency>
20.             <groupId>org.apache.flink</groupId>
21.             <artifactId>flink-java</artifactId>
22.             <version>${flink.version}</version>
23.         </dependency>
24.         <dependency>
25.             <groupId>org.apache.flink</groupId>
26.             <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
27.             <version>${flink.version}</version>
28.         </dependency>
29.         <dependency>
30.             <groupId>org.apache.flink</groupId>
31.             <artifactId>flink-clients_${scala.binary.version}</artifactId>
32.             <version>${flink.version}</version>
33.         </dependency>
34.         <dependency>
35.             <groupId>org.apache.flink</groupId>
36.             <artifactId>flink-connector-kafka_2.11</artifactId>
37.             <version>${flink.version}</version>
38.         </dependency>
39.         <dependency>
40.             <groupId>org.apache.kafka</groupId>
41.             <artifactId>kafka-clients</artifactId>
42.             <version>3.0.0</version>
43.         </dependency>
44.         <dependency>
45.             <groupId>org.apache.flink</groupId>
46.             <artifactId>flink-connector-redis_2.11</artifactId>
47.             <version>1.1.0</version>
48.         </dependency>
50.     </dependencies>
51. </project>

复制代码

KafkaProducer.java

生产数据存入到Kafka这种

1. package icu.wzk.demo05;
3. import org.apache.kafka.clients.producer.KafkaProducer;
4. import org.apache.kafka.clients.producer.Producer;
5. import org.apache.kafka.clients.producer.ProducerRecord;
7. import java.util.Properties;
9. public class TestKafkaProducer {
11.     public static void main(String[] args) throws InterruptedException {
12.         Properties props = new Properties();
13.         props.put("bootstrap.servers", "0.0.0.0:9092");
14.         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
15.         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
16.         Producer<String, String> producer = new KafkaProducer<>(props);
17.         for (int i = 0; i < 500; i++) {
18.             String key = "key-" + i;
19.             String value = "value-" + i;
20.             ProducerRecord<String, String> record = new ProducerRecord<>("test", key, value);
21.             producer.send(record);
22.             System.out.println("send: " + key);
23.             Thread.sleep(200);
24.         }
25.         producer.close();
26.     }
28. }

复制代码

StartApp

Flink消费Kafka，盘算后写入到Redis中。
FlinkJedisPoolConfig

毗连池的配置

MyRedisMapper

自定义的Mapper，需要实现RedisMapper

完整代码

1. package icu.wzk.demo05;
4. import org.apache.flink.api.common.functions.MapFunction;
5. import org.apache.flink.api.common.serialization.SimpleStringSchema;
6. import org.apache.flink.api.java.tuple.Tuple2;
7. import org.apache.flink.streaming.api.datastream.DataStreamSource;
8. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
10. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
11. import org.apache.flink.streaming.connectors.redis.RedisSink;
12. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
13. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
14. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
15. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
17. import java.util.Properties;
19. public class StartApp {
21.     private static final String KAFKA_SERVER = "0.0.0.0:9092";
23.     private static final Integer KAFKA_PORT = 9092;
25.     private static final String KAFKA_TOPIC = "test";
27.     private static final String REDIS_SERVER = "0.0.0.0";
29.     private static final Integer REDIS_PORT = 6379;
31.     public static void main(String[] args) throws Exception {
32.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
33.         Properties properties = new Properties();
34.         properties.setProperty("bootstrap.servers", String.format("%s:%d", KAFKA_SERVER, KAFKA_PORT));
35.         FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(KAFKA_TOPIC, new SimpleStringSchema(), properties);
36.         DataStreamSource<String> data = env.addSource(consumer);
38.         SingleOutputStreamOperator<Tuple2<String, String>> wordData = data.map(new MapFunction<String, Tuple2<String, String>>() {
39.             @Override
40.             public Tuple2<String, String> map(String value) throws Exception {
41.                 return new Tuple2<>("l_words", value);
42.             }
43.         });
45.         FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig
46.                 .Builder()
47.                 .setHost(REDIS_SERVER)
48.                 .setPort(REDIS_PORT)
49.                 .build();
50.         RedisSink<Tuple2<String, String>> redisSink = new RedisSink<>(conf, new MyRedisMapper());
51.         wordData.addSink(redisSink);
52.         env.execute();
53.     }
55.     public static class MyRedisMapper implements RedisMapper<Tuple2<String,String>> {
57.         @Override
58.         public RedisCommandDescription getCommandDescription() {
59.             return new RedisCommandDescription(RedisCommand.LPUSH);
60.         }
62.         @Override
63.         public String getKeyFromData(Tuple2<String,String> data) {
64.             return data.f0;
65.         }
67.         @Override
68.         public String getValueFromData(Tuple2<String,String> data) {
69.             return data.f1;
70.         }
71.     }
73. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。