Kafka相關外部系統整合
一、集成 Flume
Flume 是一個在大資料開發中非常常用的組件,
可以用于 Kafka 的生產者,也可以用于Flume 的消費者,

1、Flume 生產者

- 啟動 kafka 集群
zk.sh start
kf.sh start
- 啟動 kafka 消費者
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
- 配置 Flume
在 hadoop102 節點的 Flume 的 job 目錄下創建 file_to_kafka.conf
mkdir jobs
vim jobs/file_to_kafka.conf
組態檔內容如下:
# 1 組件定義
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 2 配置 source
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /opt/module/applog/app.*
a1.sources.r1.positionFile =
/opt/module/flume/taildir_position.json
# 3 配置 channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 4 配置 sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.bootstrap.servers =
hadoop102:9092,hadoop103:9092,hadoop104:9092
a1.sinks.k1.kafka.topic = first
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
# 5 拼接組件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
- 啟動 Flume
bin/flume-ng agent -c conf/ -n a1 -f jobs/file_to_kafka.conf &
- 向/opt/module/applog/app.log 里追加資料,查看 kafka 消費者消費情況
mkdir applog
echo hello >>/opt/module/applog/app.log
- 觀察 kafka 消費者,能夠看到消費的 hello 資料
2、Flume 消費者

配置 Flume:
在 hadoop102 節點的 Flume 的/opt/module/flume/jobs 目錄下創建 kafka_to_file.conf
vim kafka_to_file.conf
組態檔內容如下
# 1 組件定義
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 2 配置 source
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.batchSize = 50
a1.sources.r1.batchDurationMillis = 200
a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092
a1.sources.r1.kafka.topics = first
a1.sources.r1.kafka.consumer.group.id = custom.g.id
# 3 配置 channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 4 配置 sink
a1.sinks.k1.type = logger
# 5 拼接組件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
啟動 Flume:
bin/flume-ng agent -c conf/ -n a1 -f jobs/kafka_to_file.conf -Dflume.root.logger=INFO,console
啟動 kafka 生產者:
bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first
并輸入資料,例如:hello world
觀察控制臺輸出的日志
二、集成 Flink
Flink 是一個在大資料開發中非常常用的組件,
可以用于 Kafka 的生產者,也可以用于Flink 的消費者,

1、Flink 環境準備
(1)創建一個 maven 專案 flink-kafka
(2)添加組態檔
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.13.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.12</artifactId>
<version>1.13.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.12</artifactId>
<version>1.13.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.12</artifactId>
<version>1.13.0</version>
</dependency>
</dependencies>
(3)將 log4j.properties 檔案添加到 resources 里面,就能更改列印日志的級別為 error
log4j.rootLogger=error, stdout,R
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %5p --- [%50t] %-80c(line:%5L) : %m%n
log4j.appender.R=org.apache.log4j.RollingFileAppender
log4j.appender.R.File=../log/agent.log
log4j.appender.R.MaxFileSize=1024KB
log4j.appender.R.MaxBackupIndex=1
log4j.appender.R.layout=org.apache.log4j.PatternLayout
log4j.appender.R.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %5p --- [%50t] %-80c(line:%6L) : %m%n
(4)在 java 檔案夾下創建包名為 com.atguigu.flink
2、Flink 生產者
(1)在 com.atguigu.flink 包下創建 java 類:FlinkKafkaProducer1
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;
import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.ArrayList;
import java.util.Properties;
public class FlinkKafkaProducer1 {
public static void main(String[] args) throws Exception {
// 0 初始化 flink 環境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(3);
// 1 讀取集合中資料
ArrayList<String> wordsList = new ArrayList<>();
wordsList.add("hello");
wordsList.add("world");
DataStream<String> stream = env.fromCollection(wordsList);
// 2 kafka 生產者配置資訊
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
// 3 創建 kafka 生產者
FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>("first",new SimpleStringSchema(),properties);
// 4 生產者和 flink 流關聯
stream.addSink(kafkaProducer);
// 5 執行
env.execute();
}
}
(2)啟動 Kafka 消費者
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
(3)執行 FlinkKafkaProducer1 程式,觀察 kafka 消費者控制臺情況
3、Flink 消費者
(1)在 com.atguigu.flink 包下創建 java 類:FlinkKafkaConsumer1
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Properties;
public class FlinkKafkaConsumer1 {
public static void main(String[] args) throws Exception {
// 0 初始化 flink 環境
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(3);
// 1 kafka 消費者配置資訊
Properties properties = new Properties();
properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
"hadoop102:9092");
// 2 創建 kafka 消費者
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
"first",
new SimpleStringSchema(),
properties
);
// 3 消費者和 flink 流關聯
env.addSource(kafkaConsumer).print();
// 4 執行
env.execute();
}
}
(2)啟動 FlinkKafkaConsumer1 消費者
(3)啟動 kafka 生產者
bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first
(4)觀察 IDEA 控制臺資料列印
三、集成 SpringBoot
SpringBoot 是一個在 JavaEE 開發中非常常用的組件,
可以用于 Kafka 的生產者,也可以用于 SpringBoot 的消費者,

- 1)在 IDEA 中安裝 lombok 插件
在 Plugins 下搜索 lombok 然后在線安裝即可,安裝后注意重啟

- 2 )SpringBoot 環境準備
(1)創建一個 Spring Initializr

(2)專案名稱 springboot

(3)添加專案依賴




(4)檢查自動生成的組態檔
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.6.1</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.atguigu</groupId>
<artifactId>springboot</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>springboot</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<excludes>
<exclude>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>
</project>
1、SpringBoot 生產者
(1)修改 SpringBoot 核心組態檔 application.propeties, 添加生產者相關資訊
# 應用名稱
spring.application.name=atguigu_springboot_kafka
# 指定 kafka 的地址
spring.kafka.bootstrap-servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
#指定 key 和 value 的序列化器
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
(2)創建 controller 從瀏覽器接收資料, 并寫入指定的 topic
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class ProducerController {
// Kafka 模板用來向 kafka 發送資料
@Autowired
KafkaTemplate<String, String> kafka;
@RequestMapping("/atguigu")
public String data(String msg) {
kafka.send("first", msg);
return "ok";
}
}
(3)在瀏覽器中給/atguigu 介面發送資料
http://localhost:8080/atguigu?msg=hello
2、SpringBoot 消費者
(1)修改 SpringBoot 核心組態檔 application.propeties
# =========消費者配置開始=========
# 指定 kafka 的地址
spring.kafka.bootstrap-
servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
# 指定 key 和 value 的反序列化器
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#指定消費者組的 group_id
spring.kafka.consumer.group-id=atguigu
# =========消費者配置結束=========
(2)創建類消費 Kafka 中指定 topic 的資料
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
@Configuration
public class KafkaConsumer {
// 指定要監聽的 topic
@KafkaListener(topics = "first")
public void consumeTopic(String msg) { // 引數 : 收到的 value
System.out.println(" 收到的資訊: " + msg);
}
}
(3)向 first 主題發送資料
bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first
四、集成 Spark
Spark 是一個在大資料開發中非常常用的組件,
可以用于 Kafka 的生產者,也可以用于Spark 的消費者,

1、Spark 環境準備

log4j.rootLogger=error, stdout,R
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %5p --- [%50t] %-80c(line:%5L) : %m%n log4j.appender.R=org.apache.log4j.RollingFileAppender
log4j.appender.R.File=../log/agent.log
log4j.appender.R.MaxFileSize=1024KB
log4j.appender.R.MaxBackupIndex=1
log4j.appender.R.layout=org.apache.log4j.PatternLayout
log4j.appender.R.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %5p --- [%50t] %-80c(line:%6L) : %m%n
2、Spark 生產者
(1)在 com.atguigu.spark 包下創建 scala Object:SparkKafkaProducer

(2)啟動 Kafka 消費者
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
(3)執行 SparkKafkaProducer 程式,觀察 kafka 消費者控制臺情況
3、Spark 消費者
(1)添加組態檔
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.12</artifactId>
<version>3.0.0</version>
</dependency>
</dependencies>
(2)在 com.atguigu.spark 包下創建 scala Object:SparkKafkaConsumer


轉載請註明出處,本文鏈接:https://www.uj5u.com/qita/437042.html
標籤:其他
上一篇:kafka 集群服役新節點
