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）添加配置文件
org.apache.flink flink-java 1.13.0 org.apache.flink flink-streaming-java_2.12 1.13.0 org.apache.flink flink-clients_2.12 1.13.0 org.apache.flink flink-connector-kafka_2.12 1.13.0
（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 wordsList = new ArrayList<>();
wordsList.add("hello");
wordsList.add("world");
DataStream stream = env.fromCollection(wordsList);
// 2 kafka 生产者配置信息
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
// 3 创建 kafka 生产者
FlinkKafkaProducer 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 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）检查自动生成的配置文件
1、SpringBoot 生产者4.0.0 org.springframework.boot spring-boot-starter-parent 2.6.1 com.atguigu springboot 0.0.1-SNAPSHOT springboot Demo project for Spring Boot 1.8 org.springframework.boot spring-boot-starter-web org.springframework.kafka spring-kafka org.projectlombok lombok true org.springframework.boot spring-boot-starter-test test org.springframework.kafka spring-kafka-test test org.springframework.boot spring-boot-maven-plugin org.projectlombok lombok
（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 kafka;
@RequestMapping("/atguigu")
public String data(String msg) {
kafka.send("first", msg);
return "ok";
}
}
（3）在浏览器中给/atguigu 接口发送数据
http://localhost:8080/atguigu?msg=hello
（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 的消费者。
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）添加配置文件
org.apache.spark spark-streaming-kafka-0-10_2.12 3.0.0 org.apache.spark spark-core_2.12 3.0.0 org.apache.spark spark-streaming_2.12 3.0.0
（2）在 com.atguigu.spark 包下创建 scala Object：SparkKafkaConsumer
