当前位置：首页 > news >正文

kafka与flume的整合、spark-streaming

news 来源：原创 2025/4/26 6:05:42

kafka与flume的整合

前期配置完毕，开启集群

需求1：

利用flume监控某目录中新生成的文件，将监控到的变更数据发送给kafka，kafka将收到的数据打印到控制台（三个node01中运行）

1.在kafka中建立topic

kafka-topics.sh --create --zookeeper node01:2181,node02:2181,node03:2181 --topic testTopic --partitions 3 --replication-factor 3

2.启动flume（第一个nide01）

flume-ng agent -c /opt/software/flume/conf/ -f /opt/software/flume/conf/flume-kafka.conf -n a1

3.启动kafka消费者（第二个node01）

kafka-console-consumer.sh --topic testTopic --bootstrap-server node01:9092,node02:9092,node03:9092 --from-beginning

4.测试（第三个node01）

进入到指定路径下（cd /root/flume-kafka/），输入测试数据

返回到kafka消费者，可以看到数据产生

需求2：

Kafka生产者生成的数据利用Flume进行采集，将采集到的数据打印到Flume的控制台上

1.启动kafka生产者

kafka-console-producer.sh --broker-list node01:9092,node02:9092,node03:9092 --topic testTopic

2.启动flume

flume-ng agent -c /opt/software/flume/conf/ -f /opt/software/flume/conf/kafka-flume.conf -n a1 -Dflume.root.logger=INFO,console

3.在生产者中写入数据

4.在flume中采集到数据

spark-streaming

DStream转换

DStream 上的操作与 RDD 的类似，分为 Transformations（转换）和 Output Operations（输出）两种

Transform

Transform 允许 DStream 上执行任意的 RDD-to-RDD 函数。即使这些函数并没有在 DStream的 API 中暴露出来，通过该函数可以方便的扩展 Spark API。该函数每一批次调度一次。其实也就是对 DStream 中的 RDD 应用转换。

idea中的

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
// 导入 RDD 类型
import org.apache.spark.rdd.RDDobject Transform {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster("local[*]").setAppName("transform")val ssc = new StreamingContext(sparkConf, Seconds(3))val lineDStream: ReceiverInputDStream[String] = ssc.socketTextStream("node01", 9999)val wordAndCountDStream: DStream[(String, Int)] = lineDStream.transform(rdd => {val words: RDD[String] = rdd.flatMap(_.split(" "))val wordAndOne: RDD[(String, Int)] = words.map((_, 1))val value: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)value})wordAndCountDStream.print()ssc.start()ssc.awaitTermination()}
}

虚拟机中的

join

两个流之间的 join 需要两个流的批次大小一致，这样才能做到同时触发计算。计算过程就是对当前批次的两个流中各自的 RDD 进行 join，与两个 RDD 的 join 效果相同

idea中的

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.rdd.RDDobject join {def main(args: Array[String]): Unit = {// 创建 SparkConf 对象，设置运行模式为本地多线程，应用名为 joinval sparkConf = new SparkConf().setMaster("local[*]").setAppName("join")// 创建 StreamingContext 对象，设置批处理间隔为 3 秒val ssc = new StreamingContext(sparkConf, Seconds(3))// 从 node01 的 9999 端口接收文本流val lineDStream1: ReceiverInputDStream[String] = ssc.socketTextStream("node01", 9999)// 从 node02 的 8888 端口接收文本流val lineDStream2: ReceiverInputDStream[String] = ssc.socketTextStream("node02", 8888)// 将 lineDStream1 中的每行文本拆分为单词，并映射为 (单词, 1) 的键值对val wordToOneDStream: DStream[(String, Int)] = lineDStream1.flatMap(_.split(" ")).map((_, 1))// 将 lineDStream2 中的每行文本拆分为单词，并映射为 (单词, "a") 的键值对val wordToADstream: DStream[(String, String)] = lineDStream2.flatMap(_.split(" ")).map((_, "a"))// 对两个 DStream 进行 join 操作，结果为 (单词, (1, "a")) 的键值对val joinDStream: DStream[(String, (Int, String))] = wordToOneDStream.join(wordToADstream)// 打印 join 操作后的结果joinDStream.print()// 启动 StreamingContextssc.start()// 等待计算终止ssc.awaitTermination()}
}

虚拟机中的

node01

nc -lk 9999

node02

nc -lk 8888