以SparkStreaming + Kafka 实现

假定用户有某个周末网民网购停留时间的日志文本，基于某些业务要求，要求开发

Spark应用程序实现如下功能：

1、实时统计连续网购时间超过半个小时的女性网民信息。

2、周末两天的日志文件第一列为姓名，第二列为性别，第三列为本次停留时间，单

位为分钟，分隔符为“,”。

数据：

log1.txt：周六网民停留日志

LiuYang,female,20

YuanJing,male,10

GuoYijun,male,5

CaiXuyu,female,50

Liyuan,male,20

FangBo,female,50

LiuYang,female,20

YuanJing,male,10

GuoYijun,male,50

CaiXuyu,female,50

FangBo,female,60

log2.txt：周日网民停留日志

LiuYang,female,20

YuanJing,male,10

CaiXuyu,female,50

FangBo,female,50

GuoYijun,male,5

CaiXuyu,female,50

Liyuan,male,20

CaiXuyu,female,50

FangBo,female,50

LiuYang,female,20

YuanJing,male,10

FangBo,female,50

GuoYijun,male,50

CaiXuyu,female,50

FangBo,female,60

实现步骤

一、pom.xml

<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.xu.sparktest1</groupId>
    <artifactId>sparktest1</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <spark.version>2.1.0</spark.version>
        <scala.version>2.11</scala.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>
    </dependencies>

    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>

        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.5.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>

            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.0</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

二、scala代码

package com.xu

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{HashPartitioner, SparkConf}

object SparkStreamTest5 {
  def main(args: Array[String]): Unit = {

    val sparkConf=new SparkConf().setAppName("SparkHomeWork2").setMaster("local[2]")

    val ssc=new StreamingContext(sparkConf,Seconds(5))
    ssc.sparkContext.setLogLevel("WARN")
    ssc.checkpoint(".")

    //创建连接kafka的参数
    val brokeList = "node01:9092,node02:9092,node03:9092"
    val zk = "node01:2181/kafka"
    val sourceTopic = "sparkhomework-test4"
    val consumerGroup = "sparkhomework2"

    val topicMap = sourceTopic.split(",").map((_, 1.toInt)).toMap

    val lines = KafkaUtils.createStream(ssc, zk, consumerGroup, topicMap).map(_._2)
    val results = lines.flatMap(_.split(" ")).filter(_.contains("female"))

    val femaleData: DStream[(String, Int)] = results.map { line =>
        val t = line.split(',')
        (t(0), t(2).toInt)
      }.reduceByKey(_ + _)
    //筛选出时间大于两个小时的女性网民信息，并输出
    val date = femaleData.updateStateByKey(updateFunction, new HashPartitioner(ssc.sparkContext.defaultParallelism), true).filter(line => line._2 > 120)
    date.print()
    ssc.start()
    ssc.awaitTermination()
  }

  val updateFunction = (iter: Iterator[(String, Seq[Int], Option[Int])]) => {
    iter.flatMap { case (x, y, z) => Some(y.sum + z.getOrElse(0)).map(v => (x, v)) }
  }
}

三启动kafka：

kafka集群启动，必须先启动zookeeper：

启动zookeeper

三台服务器启动zookeeper，三台机器都执行以下命令启动zookeeper:

cd  /export/servers/zookeeper-3.4.5-cdh5.14.0
bin/zkServer.sh start

bin/zkServer.sh  status

启动kafka：在三台服务上执行：

cd /export/servers/kafka_2.11-0.10.0.0
nohup bin/kafka-server-start.sh config/server.properties 2>&1 &

停止kafka，在三台服务上执行：

cd /export/servers/kafka_2.11-0.10.0.0
bin/kafka-server-stop.sh

四数据准备

将上述log进行压缩，并以空格进行进行分割，如下：

LiuYang,female,20 YuanJing,male,10 GuoYijun,male,5 CaiXuyu,female,50 Liyuan,male,20 FangBo,female,50 LiuYang,female,20 YuanJing,male,10 GuoYijun,male,50 CaiXuyu,female,50 FangBo,female,60

LiuYang,female,20 YuanJing,male,10 CaiXuyu,female,50 FangBo,female,50 GuoYijun,male,5 CaiXuyu,female,50 Liyuan,male,20 CaiXuyu,female,50 FangBo,female,50 LiuYang,female,20 YuanJing,male,10 FangBo,female,50 GuoYijun,male,50 CaiXuyu,female,50 FangBo,female,60

五、kafka创建Topic

cd /export/servers/kafka_2.11-0.10.0.0

--使用命令创建 Topic
bin/kafka-topics.sh --create --topic sparkhomework-test4 --replication-factor 1 --partitions 3 --zookeeper node01:2181/kafka

--开启 Producer
bin/kafka-console-producer.sh --broker-list node01:9092,node02:9092,node03:9092 --topic sparkhomework-test4

SparkStreaming完整demo（kafka版）-累积个数（updateStateByKey）以SparkStreaming + Kafka 实现实现步骤

注意：我这里kafka的路径为：

node01:2181/kafka

，这个是在config/server.properties中配置的，

SparkStreaming完整demo（kafka版）-累积个数（updateStateByKey）以SparkStreaming + Kafka 实现实现步骤

六测试

启动scala的main方法，然后在控制台依次输入 log1，log2：

SparkStreaming完整demo（kafka版）-累积个数（updateStateByKey）以SparkStreaming + Kafka 实现实现步骤

结果：

SparkStreaming完整demo（kafka版）-累积个数（updateStateByKey）以SparkStreaming + Kafka 实现实现步骤

如果在LInux控制台输入log时，程序打印，并没有出现红色部分的内容，表示topic配置问题，并没有正确连接！！！

SparkStreaming完整demo（kafka版）-累积个数（updateStateByKey）以SparkStreaming + Kafka 实现实现步骤

以SparkStreaming + Kafka 实现

实现步骤

一、pom.xml

二、scala代码

三启动kafka：

启动zookeeper

启动kafka：在三台服务上执行：

停止kafka，在三台服务上执行：

四数据准备

五、kafka创建Topic

六测试

继续阅读

用c++ 连接kafka我所踩过的坑(Connection refused || desired partition does not exist)

Linux 安装kafka的库librdkafka

librdkafka安装步骤

Kafka：Topic概念与API介绍

Doris SQL日志审计部署，以及sql收集输出kafka，后续血缘分析

Flink Kafka Doris实战demo

延云行业搜索数据库在大数据生态中位置和重要性大数据的挑战大数据技术的现状延云行业搜索数据库

Spark在windows环境里跑时报错找不到org.apache.hadoop.fs.FSDataInputStream

Kafka 和 EMS 消息批量 ack 的实现

Spark流式分析系统实现流式实时日志分析系统

Spring Cloud整合Sleuth，当请求完成后，Zipkin没有链路信息

Scala和Java二种方式实战Spark Streaming开发

Spark基础:Spark简介及特点,运行模式,安装Spark,Driver与Executor,Local模式,Standalone模式,Yarn模式,Mesos模式,WordCount案例,HA配置第1章 Spark概述第2章 Spark运行模式第3章案例实操

Spark实现wordcount

大数据排错SparkSpark集群启动时候，JAVA_HOME is not sethadoop集群，某台服务器jps无任何输出IDEAkafkahadoopspark sqlfile permissionsIDEA本地测试 - OutOfMemoryError: GC overhead limit exceededhdfs负载均衡

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

SparkStreaming完整demo（kafka版）-累积个数（updateStateByKey）以SparkStreaming + Kafka 实现实现步骤

以SparkStreaming + Kafka 实现

实现步骤

一 、pom.xml

二 、scala代码

三 启动kafka：

启动zookeeper

启动kafka：在三台服务上执行：

停止kafka，在三台服务上执行：

四 数据准备

五、kafka创建Topic

六 测试

继续阅读

一、pom.xml

二、scala代码

三启动kafka：

四数据准备

六测试