kafka的pom文件_Flink的sink实战之二：kafka

2. 在控制台消费test006的消息&＃xff0c;参考命令&＃xff1a;
&＃96;&＃96;&＃96;shell
./kafka-console-consumer.sh
--bootstrap-server 127.0.0.1:9092
--topic test006

2. 根据提示&＃xff0c;groupid输入com.bolingcavalry&＃xff0c;artifactid输入flinksinkdemo&＃xff0c;即可创建一个maven工程&＃xff1b;
3. 在pom.xml中增加kafka依赖库&＃xff1a;
&＃96;&＃96;&＃96;xml
org.apache.flinkflink-connector-kafka_2.111.9.0


private String topic;public ProducerStringSerializationSchema(String topic) {super();this.topic &＃61; topic;
}&＃64;Override
public ProducerRecord serialize(String element, Long timestamp) {return new ProducerRecord(topic, element.getBytes(StandardCharsets.UTF_8));
}

2. 创建任务类KafkaStrSink&＃xff0c;请注意FlinkKafkaProducer对象的参数&＃xff0c;FlinkKafkaProducer.Semantic.EXACTLY_ONCE表示严格一次&＃xff1a;
&＃96;&＃96;&＃96;java
package com.bolingcavalry.addsink;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;public class KafkaStrSink {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();//并行度为1env.setParallelism(1);Properties properties &＃61; new Properties();properties.setProperty("bootstrap.servers", "192.168.50.43:9092");String topic &＃61; "test006";FlinkKafkaProducer producer &＃61; new FlinkKafkaProducer<>(topic,new ProducerStringSerializationSchema(topic),properties,FlinkKafkaProducer.Semantic.EXACTLY_ONCE);//创建一个List&＃xff0c;里面有两个Tuple2元素List list &＃61; new ArrayList<>();list.add("aaa");list.add("bbb");list.add("ccc");list.add("ddd");list.add("eee");list.add("fff");list.add("aaa");//统计每个单词的数量env.fromCollection(list).addSink(producer).setParallelism(4);env.execute("sink demo : kafka str");}
}

package com.bolingcavalry.addsink;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;
import javax.annotation.Nullable;public class ObjSerializationSchema implements KafkaSerializationSchema> {private String topic;private ObjectMapper mapper;public ObjSerializationSchema(String topic) {super();this.topic &＃61; topic;}&＃64;Overridepublic ProducerRecord serialize(Tuple2 stringIntegerTuple2, &＃64;Nullable Long timestamp) {byte[] b &＃61; null;if (mapper &＃61;&＃61; null) {mapper &＃61; new ObjectMapper();}try {b&＃61; mapper.writeValueAsBytes(stringIntegerTuple2);} catch (JsonProcessingException e) {// 注意&＃xff0c;在生产环境这是个非常危险的操作&＃xff0c;// 过多的错误打印会严重影响系统性能&＃xff0c;请根据生产环境情况做调整e.printStackTrace();}return new ProducerRecord(topic, b);}
}

package com.bolingcavalry.addsink;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;public class KafkaObjSink {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();//并行度为1env.setParallelism(1);Properties properties &＃61; new Properties();//kafka的broker地址properties.setProperty("bootstrap.servers", "192.168.50.43:9092");String topic &＃61; "test006";FlinkKafkaProducer> producer &＃61; new FlinkKafkaProducer<>(topic,new ObjSerializationSchema(topic),properties,FlinkKafkaProducer.Semantic.EXACTLY_ONCE);//创建一个List&＃xff0c;里面有两个Tuple2元素List> list &＃61; new ArrayList<>();list.add(new Tuple2("aaa", 1));list.add(new Tuple2("bbb", 1));list.add(new Tuple2("ccc", 1));list.add(new Tuple2("ddd", 1));list.add(new Tuple2("eee", 1));list.add(new Tuple2("fff", 1));list.add(new Tuple2("aaa", 1));//统计每个单词的数量env.fromCollection(list).keyBy(0).sum(1).addSink(producer).setParallelism(4);env.execute("sink demo : kafka obj");}
}