安装 kafka 需要先安装 jdk。一、下载官网：http://kafka.apache.org/downloads.html       二、安装 安装参考：https://segmentfault.com/a/1190000012990954 1. 解压安装（我的安装目录/opt/kafka/） # t…

vi kafka-server-stop.sh   把PIDS=$(ps ax | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $1}')这行代码修改为 PIDS=$(jps -lm | grep -i 'kafka.Kafka' | awk '…

主要思路：SparkContext由spring容器管理，在spring容器启动完毕后，执行spark-streaming-kafka，获取数据并处理。 1.spring容器中初始化SparkContext，代码片段如下： @Bean @ConditionalOnMissingBean(SparkConf.class) public Sp…

官网介绍 http://spark.apache.org/docs/2.3.0/streaming-kafka-0-10-integration.html#creating-a-direct-stream   案例pom.xml依赖 <dependency> <groupId>org.apache.spa…

①进入到kafka文件夹中修改配置文件：vim config/server.properties         ②启动zookeeper： bin/zookeeper-server-start.sh config/zookeeper.properties 端口2181是ZooKeeper的默认端口…

Kafka的消息消费是以消费的group为单位; 同属一个group中的多个consumer分别消费topic的不同partition; 同组内consumer的变化, partition变化, coordinator的变化都会引发balance; 消费的offset的提交 Kafka wiki: Kafka Detailed Consu…

前面有说道spark-streaming的简单demo，也有说到kafka成功跑通的例子，这里就结合二者，也是常用的使用之一。 1.相关组件版本 首先确认版本，因为跟之前的版本有些不一样，所以才有必要记录下，另外仍然没有使用scala,使用java8,spark 2.0.0,kafka 0.10。 2.引入maven包 网上找了一些结合的…

1.定义 Kafka中的每个partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到partition中。partition中的每个消息都有一个连续的序号，用于partition唯一标识一条消息。 Offset记录着下一条将要发送给Consumer的消息的序号。 流处理系统常见的三种语义： 最多一次 每个记录要么处理一…

spark streaming拉取kafka数据， 结合sparkSql dataframe hive存储计算，输出到mysql. 数据清洗过程比较复杂，没办法，上游给的屡一样的数据，正则去解析并全量按时间取最新一条去重。 每天kafka数据5千万条。1分钟要刷一次，看上去还可以满足。只有屡一样去堆代码了。 package biRepor…

背景： kafka流量在800M/s,前任留下的程序大量数据丢失，且逻辑生成复杂，查询hive直接奔溃，优化从两方面，程序优化及小文件合并（生成结果产生大量小文件） 程序直接上代码，啥也不说了 程序 def main(args: Array[String]): Unit = { val sdf = new SimpleDateFormat…