Apache Flink：使用Apache Kafka作為Sink的簡單demo(資料結果存放地方)

1. 聲明

目前内容主要為測試和使用Flink，将資料讀取處理後放入到kafka的topic中

主要内容：

1. 使用Flink讀取文本内容
2. 過濾讀取的内容
3. 将讀取的内容放入kafka中

2.基本demo

1.pom依賴

<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<flink.version>1.13.0</flink.version>
		<target.java.version>1.8</target.java.version>
		<scala.binary.version>2.11</scala.binary.version>
		<maven.compiler.source>${target.java.version}</maven.compiler.source>
		<maven.compiler.target>${target.java.version}</maven.compiler.target>
		<log4j.version>2.12.1</log4j.version>
	</properties>

	<dependencies>
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-walkthrough-common_${scala.binary.version}</artifactId>
			<version>${flink.version}</version>
		</dependency>

		<!-- This dependency is provided, because it should not be packaged into 
			the JAR file. -->
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
			<version>${flink.version}</version>
			<!-- <scope>provided</scope> -->
		</dependency>
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-clients_${scala.binary.version}</artifactId>
			<version>${flink.version}</version>
			<!-- <scope>provided</scope> -->
		</dependency>
		<!-- 直接導入需要的flink到kafka的連接配接器 -->
		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-connector-kafka_2.12</artifactId>
			<version>1.13.0</version>
		</dependency>
		<dependency>
			<groupId>org.apache.logging.log4j</groupId>
			<artifactId>log4j-slf4j-impl</artifactId>
			<version>${log4j.version}</version>
			<scope>runtime</scope>
		</dependency>
		<dependency>
			<groupId>org.apache.logging.log4j</groupId>
			<artifactId>log4j-api</artifactId>
			<version>${log4j.version}</version>
			<scope>runtime</scope>
		</dependency>
		<dependency>
			<groupId>org.apache.logging.log4j</groupId>
			<artifactId>log4j-core</artifactId>
			<version>${log4j.version}</version>
			<scope>runtime</scope>
		</dependency>
	</dependencies>
           

2.開啟一個指令行消費者

3.開始編寫demo

import java.util.Properties;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

/**
 * 
 * @author hy
 * @createTime 2021-05-23 10:38:40
 * @description 目前内容主要将Apache Flink中處理的内容放入目前的Apache Kafka中(将Kafka作為資料輸出的地點)
 *
 */
public class LocalRunningToKafkaSaveTest {
	public static void main(String[] args) {
		// 采用本地模式
		StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

		// 設定資料來源為集合資料
		String topic = "test-events"; // 設定監聽的主題
		Properties props = new Properties();
		props.setProperty("bootstrap.servers", "192.168.1.101:9092");
		props.setProperty("group.id", "test");
		String filePath = "D:\\eclipse-workspace\\Apache-Flink-Start\\resources\\abc.txt";
		DataStream<String> txtStream = env.readTextFile(filePath);

		// 這裡可以進行處理操作(但是結果需要指派)
		txtStream = txtStream.filter(new FilterFunction<String>() {
			@Override
			public boolean filter(String value) throws Exception {
				// TODO Auto-generated method stub
				// 傳回資料中不帶有19的資料
				if(!value.contains("19")) {
					return true;
				}
				return false;
			}
		});
		
		System.out.println("處理後的資料=====>");
		txtStream.print();
		System.out.println("<=====處理後的資料");
		
		// 将讀取的文本類型的資料注入到kafka的這個topic中
		FlinkKafkaProducer<String> producer=new FlinkKafkaProducer<String>(topic, new SimpleStringSchema(),props);
		txtStream.addSink(producer);
		System.out.println("将目前的資料處理後放入kafka的test-events這個topic成功！");
		try {
			// 最後開始執行
			JobExecutionResult result = env.execute("Fraud Detection");
			if (result.isJobExecutionResult()) {
				System.out.println("執行完畢......");
			}
			System.out.println();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}

           

小心這裡的filter,以及使用的Sink為FlinkKafkaProducer(一個消息生産者)

3. 測試結果

指令行消費者結果：

4. 總結

1.由于本人沒有找到一個清空topic的消息的，隻要是訂閱了的，以前釋出的消息都會發送，并且接受到，隻有執行删除topic的指令，然後在建立才有用

./bin/kafka-topics.sh --delete --topic test-events --bootstrap-server 192.168.1.101:9092