flink-exactly-once

1.运行kafka

选择 wurstmeister/zookeeper、wurstmeister/kafka 两个镜像快速启动kafka。

1.1 docker运行kafka

View Code

docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

ip=`ifconfig en0 | sed -n '/inet/p'| sed -n '/netmask/p' | awk -F' ' '{print $2}'`
docker run  -d --name kafka \
    -p 9092:9092 \
    -e KAFKA_BROKER_ID=0 \
    -e KAFKA_ZOOKEEPER_CONNECT=$ip:2181 \
    -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://$ip:9092 \
    -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -t wurstmeister/kafka

docker run  -d --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=192.168.1.2:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.1.2:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -t wurstmeister/kafka

1.2 创建kafka主题

View Code

bin/kafka-topics.sh --create --zookeeper 192.168.1.2:2181 --replication-factor 1 --partitions 1 --topic topic-input

bin/kafka-topics.sh --create --zookeeper 192.168.1.2:2181 --replication-factor 1 --partitions 1 --topic topic-output

2.flink kafka 使用

fink kafka 版本依赖

View Code

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <java.version>1.8</java.version>
    <flink.version>1.11.1</flink.version>
    <scala.version>2.11</scala.version>
</properties>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_${scala.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

参考文档

https://flink.apache.org/features/2018/03/01/end-to-end-exactly-once-apache-flink.html
https://ververica.cn/developers/anatomy-of-flink-kafka-connector-and-exactly-once/
https://ververica.cn/developers/flink-kafka-end-to-end-exactly-once-analysis/
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/connectors/kafka.html
https://blog.csdn.net/lblblblblzdx/article/details/80548294

打赏