头歌实践教学平台 大数据编程 实训答案（三）

第一章 遍历日志数据

用 Spark 遍历日志数据

第1关：用 Spark 获得日志文件中记录总数

任务描述
本关任务：编写一个能用 Spark 操作日志文件并输出日志文件记录数的小程序。

相关知识
为了完成本关任务，你需要掌握：1.搜索查询日志的内容，2.如何用 Spark 获得日志文件，3.如何获得日志文件的记录数。

import org.apache.spark.{SparkConf, SparkContext}

object Test1 {

  def main(args: Array[String]) {

    // SparkConf包含了Spark配置的各种参数，
    // local：设置为本地运行
    // *：使用本地的所有cpu核
    // setAppName：设置本应用程序的别名（自定义）
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sou")

    // 进入Spark操作的入口
    val sc = new SparkContext(sparkConf)

    // 获得文本文件内容
    val sou = sc.textFile("/data/workspace/myshixun/projectsou1_1/src/soulog.txt")

    //********** Begin **********
    println("搜索日志文件中共有%d条记录".format(sou.count())) 

    //********** End **********


  }
}

第2关：用 Spark 获得日志文件中记录内容

任务描述
本关任务：编写一个能用 Spark 获得日志文件中记录内容的小程序。

相关知识
为了完成本关任务，你需要掌握：如何用 Spark 获得日志文件中记录内容。

import org.apache.spark.{SparkConf, SparkContext}

object Test2 {

  def main(args: Array[String]) {

    // SparkConf包含了Spark配置的各种参数，
    // local：设置为本地运行
    // *：使用本地的所有cpu核
    // setAppName：设置本应用程序的别名（自定义）
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sou")

    // 进入Spark操作的入口
    val sc = new SparkContext(sparkConf)

    // 获得文本文件内容
    val sou = sc.textFile("/data/workspace/myshixun/projectsou1_1/src/soulog.txt")

    //********** Begin **********
    val rdd1 = sou.map {
      case log =>
        val logSplit = log.split("\\s")
        (logSplit(3),logSplit(4))
    }
    rdd1.collect.take(6).foreach(println(_)) 
    //********** End **********


  }
}

第二章 过滤日志数据

用 Spark 过滤日志数据

第1关：掌握用 Spark 过滤日志数据

任务描述
本关任务：编写一个能用 Spark 过滤日志数据的小程序。

相关知识
为了完成本关任务，你需要掌握：如何用 Spark 过滤日志数据。

import org.apache.spark.{SparkConf, SparkContext}

object Test1 {

  def main(args: Array[String]) {

    // SparkConf包含了Spark配置的各种参数，
    // local：设置为本地运行
    // *：使用本地的所有cpu核
    // setAppName：设置本应用程序的别名（自定义）
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sou")

    // 进入Spark操作的入口
    val sc = new SparkContext(sparkConf)

    // 获得文本文件内容
    val sou = sc.textFile("/data/workspace/myshixun/projectsou2_1/src/soulog.txt")

    //********** Begin **********
    val splitSou = sou.map(_.split("\\s"))
    val filterSou = splitSou  
      .filter(_ (3).toInt == 1)
      .filter(_ (4).toInt == 1)
    print(filterSou.count())

    //********** End **********

  }
}

第三章 聚合、排序日志数据

用 Spark 对日志数据进行排序

第1关：用 Spark 对日志数据进行排序

任务描述
本关任务：编写一个能用 Spark 对日志数据进行排序的小程序。

相关知识
为了完成本关任务，你需要掌握：如何用 Spark 对日志数据进行排序。

排序操作
要对上节课的数据进行排序操作，才能从大到小输出排行榜，
比如上节课获得的数据是：

(222,1)
(111,3)
(333,2)

格式是（用户id，查询次数），

现在要将这些数据按照查询次数的从大到小进行排序，也就是降序排序，代码如下：

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sou")
    val sou = sc.textFile("src/soulog2.txt")
    val splitSou = sou.map(_.split("\\s"))
    val result=splitSou.map(x => (x(1),1))
      .reduceByKey(_+_)
    // 将之前的结果进行降序排序，输出用户查询次数的排行榜
      val sortResult=result
      .map(x => (x._2,x._1))
      .sortByKey(false)
      .map(x => (x._2,x._1))
    sortResult.collect().foreach(println(_)) 

import org.apache.spark.{SparkConf, SparkContext}

object Test1 {

  def main(args: Array[String]) {

    // SparkConf包含了Spark配置的各种参数，
    // local：设置为本地运行
    // *：使用本地的所有cpu核
    // setAppName：设置本应用程序的别名（自定义）
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sou")

    // 进入Spark操作的入口
    val sc = new SparkContext(sparkConf)

    // 获得文本文件内容
    val sou = sc.textFile("/data/workspace/myshixun/projectsou4_1/src/soulog.txt")

    //\s代表正则表达式中的一个空白字符（可能是空格、制表符、其他空白）
    //分割后，输出总记录数
    val splitSou = sou.map(_.split("\\s"))

    // 日志文件中总共有10000条记录
    println(splitSou.count())

    //只查询在返回结果中的排名和用户点击的顺序号都为1的记录，
    // 之前已经分隔成6个部分的数据，
    // 现在我们要查询排名第1（也就是第4个部分数据）以及用户点击排名第1（也就是第5个数据）的数据
    // 可以用连续的filter方法来进行多次过滤，
    // 注意将排名值用toInt方法转换为整数
    val filterSou = splitSou
      .filter(_ (3).toInt == 1)
      .filter(_ (4).toInt == 1)

    // 获得经过以上过滤的数据，并且，将每条记录的用户id取出来，并给每条记录加一个value值为1
    val result=filterSou.map(x => (x(1),1))
    // 将相同用户的查询次数统计出来
      .reduceByKey(_+_)

    //********** Begin **********
    val sortResult=result
        // 因为我们要按key进行排序，而之前的结果的key是用户id，value是次数
        // 所以我们将原来的key和value互换位置，
        // x._1就是（key，value）的第一个元素key，x._2就是（key，value）的第二个元素value
        // 所以我们用map方法互换了key和value的位置
      .map(x => (x._2,x._1))
        // 然后按照现在的key也就是查询次数来进行排序，因为是排行榜，从大到小，所以是降序排序
      .sortByKey(false)
        //排完序后，再将排完序的数据的key和value进行互换，
      .map(x => (x._2,x._1))
    // 输出用户查询次数
    sortResult.collect().take(10).foreach(println(_)) 


    //********** End **********


  }
}

用 Spark 对日志数据进行聚合

第1关：用 Spark 对日志数据进行聚合

任务描述
本关任务：编写一个能用 Spark 对日志数据进行聚合的小程序。

相关知识
为了完成本关任务，你需要掌握：如何用 Spark 对日志数据进行聚合。

聚合操作
我们经常要对数据进行聚合操作，

比如对于以下数据：
时间       用户id   查询的词
00:00:00    111    [查询词1]
00:00:00    111    [查询词2]
00:00:00    333    [查询词3]
00:00:00    111    [查询词4]
00:00:00    222    [查询词5]
00:00:00    333    [查询词5]

我们要查询出每个用户查询的次数，
可以用以下代码来实现：

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sou")
    val sc = new SparkContext(sparkConf)
    val sou = sc.textFile("src/soulog2.txt")
    val splitSou = sou.map(_.split("\\s"))
    val result=splitSou.map(x => (x(1),1))
      .reduceByKey(_+_)
    result.collect().foreach(println(_)) 

import org.apache.spark.{SparkConf, SparkContext}

object Test1 {

  def main(args: Array[String]) {

    // SparkConf包含了Spark配置的各种参数，  
    // local：设置为本地运行  
    // *：使用本地的所有cpu核  
    // setAppName：设置本应用程序的别名（自定义）  
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sou")
    // 进入Spark操作的入口  
    val sc = new SparkContext(sparkConf)
    // 获得文本文件内容  
    val sou = sc.textFile("/data/workspace/myshixun/projectsou3_1/src/soulog.txt")
    //\s代表正则表达式中的一个空白字符（可能是空格、制表符、其他空白）  
    //分割后，输出总记录数  
    val splitSou = sou.map(_.split("\\s"))
    // 日志文件中总共有10000条记录  
    println(splitSou.count())
    //只查询在返回结果中的排名和用户点击的顺序号都为1的记录，  
    // 之前已经分隔成6个部分的数据，  
    // 现在我们要查询排名第1（也就是第4个部分数据）以及用户点击排名第1（也就是第5个数据）的数据  
    // 可以用连续的filter方法来进行多次过滤，  
    // 注意将排名值用toInt方法转换为整数  
    val filterSou = splitSou  
      .filter(_ (3).toInt == 1)  
      .filter(_ (4).toInt == 1)  

    //********** Begin **********
    val result=filterSou.map(x => (x(1),1))
      // 将相同用户的查询次数统计出来
      .reduceByKey(_+_)
    result.collect().take(5).foreach(println(_)) 

    //********** End **********


  }
}

第一章 Spark 入门

Spark Standalone 模式的安装和部署

第1关： Standalone 分布式集群搭建

任务描述
掌握 Standalone 分布式集群搭建。

相关知识
我们已经掌握了 Spark 单机版安装，那么分布式集群怎么搭建呢？ 接下来我们学习 Standalone 分布式集群搭建。

启动环境

cd /home
wrapdocker
ulimit -f 1024000
# 加载镜像
docker load -i hbase-ssh2_v1.0.tar
# 启动集群 启动失败则等一会，再次执行，直至成功
docker-compose up -d
# 新开一个命令行窗口 master         密码统一为 123456
ssh 172.18.0.2
ssh-keygen -t rsa #三下回车
# 新开一个命令行窗口 slave1
ssh 172.18.0.3
ssh-keygen -t rsa #三下回车
# 新开一个命令行窗口 slave2
ssh 172.18.0.4
ssh-keygen -t rsa #三下回车
#  在 master 复制 master、slave1、slave2 的公钥。
cat ~/.ssh/id_rsa.pub>> ~/.ssh/authorized_keys
ssh root@slave1 cat ~/.ssh/id_rsa.pub>> ~/.ssh/authorized_keys
ssh root@slave2 cat ~/.ssh/id_rsa.pub>> ~/.ssh/authorized_keys
# 分别在 slave1、slave2 复制 master 的 authorized_keys 文件。
ssh root@master cat ~/.ssh/authorized_keys>> ~/.ssh/authorized_keys
# 在第1个命令行窗口 密码为123456
scp -r /usr/local/spark-2.3.4-bin-hadoop2.7 root@172.18.0.2:/usr/local
# 在master(第2个命令行窗口) 修改配置 注意：未提示更换命令行则一直在master上执行
echo "export SPARK_HOME=/usr/local/spark-2.3.4-bin-hadoop2.7" >> /etc/profile
source /etc/profile
cd /usr/local/spark-2.3.4-bin-hadoop2.7/conf 
mv spark-env.sh.template spark-env.sh
echo "export JAVA_HOME=/usr/lib/jvm/jdk1.8.0_111" >> spark-env.sh
echo "SPARK_MASTER_WEBUI_PORT=8888" >> spark-env.sh
echo "export SPARK_MASTER_IP=master" >> spark-env.sh
mv slaves.template slaves 
vi slaves
# 修改为以下内容
master
slave1
slave2
# 分发安装包
cd /usr/local
scp -r spark-2.3.4-bin-hadoop2.7/ root@slave1:/usr/local
scp -r spark-2.3.4-bin-hadoop2.7/ root@slave2:/usr/local
# 启动集群
cd /usr/local/spark-2.3.4-bin-hadoop2.7/sbin
./start-all.sh

Spark的安装与使用

第1关：Scala 环境的安装与部署

任务描述
本关任务：安装与配置Scala开发环境。

相关知识
Scala是一种函数式面向对象语言，它融汇了许多前所未有的特性，而同时又运行于JVM之上。随着开发者对Scala的兴趣日增，以及越来越多的工具支持，无疑Scala语言将成为你手上一件必不可少的工具。

而我们将要学习的大数据框架Spark底层是使用Scala开发的，使用scala写出的代码长度是使用java写出的代码长度的1/10左右，代码实现更加简练。

所以安装与配置Scala的环境是我们在开始学习Spark之前要完成的准备工作。

接下来我们开始安装，分为三个步骤：

下载解压；
配置环境；
校验。

启动环境

mkdir /app
cd /opt/
tar -zxvf  scala-2.12.7.tgz -C /app
vi /etc/profile
SCALA_HOME=/app/scala-2.12.7
export PATH=$PATH:$SCALA_HOME/bin
source /etc/profile

第2关：Spark 环境安装

任务描述
本关任务：安装与配置Spark开发环境。

相关知识
Apache Spark是专为大规模数据处理而设计的快速通用的计算引擎。Spark是UC Berkeley AMP lab(加州大学伯克利分校的AMP实验室)所开源的类Hadoop MapReduce的通用并行框架，Spark拥有Hadoop MapReduce所具有的优点；但不同于MapReduce的是——Job中间输出结果可以保存在内存中，从而不再需要读写HDFS，因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。

本关我们来配置一个伪分布式的Spark开发环境，与配置Hadoop类似分为三个步骤：

下载解压安装包；
配置环境变量；
配置Spark环境；
校验。

cd /opt/
tar -zxvf spark-2.2.2-bin-hadoop2.7.tgz -C /app
vim /etc/profile
SPARK_HOME=/app/spark-2.2.2-bin-hadooop2.7
export PATH=$PATH:$SPARK_HOME/bin
source /etc/profile
cd /app/spark-2.2.2-bin-hadoop2.7/conf
cp spark-env.sh.template spark-env.sh
vi spark-env.sh
export JAVA_HOME=/usr/lib/jvm/jdk1.8.0_111
export SCALA_HOME=/app/scala-2.12.7
export HADOOP_HOME=/usr/local/hadoop/
export HADOOP_CONF_DIR=/usr/local/hadoop/etc/hadoop
export SPARK_MASTER_IP=127.0.0.1
export SPARK_LOCAL_IP=127.0.0.1
cd /app/spark-2.2.2-bin-hadoop2.7
./sbin/start-all.sh

第三章 SparkSQL结构化数据分析与处理

军用大数据 - 结构化数据分析与处理

第1关：Spark SQL入门

任务描述
掌握 Spark SQL 相关基础知识，完成选择题任务。

相关知识
Spark SQL 是 Spark 用来处理结构化数据的一个模块。Spark SQL 为了支持结构化数据的处理，它提供了两个编程抽象分别叫做 DataFrame 和DataSet。

1、C；2、AB

第2关：使用Spark SQL统计战斗机飞行性能

任务描述
通过飞行速度统计出战斗机飞行性能排比。

相关知识
本关使用  mySQL 统计战斗机飞行性能。

# coding=utf-8


from pyspark.sql import SparkSession

#**********Begin**********#

#创建SparkSession
spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .config("spark.sql.crossJoin.enabled", "true") \
    .master("local") \
    .getOrCreate()
    
#读取/root/jun.json中数据
df =spark.read.json("/root/jun.json")
#创建视图
df.createOrReplaceTempView("table1")
#统计出全球飞行速度排名前三的战斗机
sqlDF = spark.sql("select cast(regexp_replace(regexp_extract(`最大飞行速度`,'[\\\d,\\\.]+',0),',','') as float) as SPEED, `名称` from table1 order by SPEED desc LIMIT 3")
#保存结果
sqlDF.write.format("csv").save("/root/airspark")


#**********End**********#
spark.stop()

第3关：使用Spark SQL统计各个研发单位研制战斗机占比

任务描述
统计出各个研发单位研制战斗机占比。

相关知识
使用 Spark SQL 统计各个研发单位研制战斗机占比。

# coding=utf-8


from pyspark.sql import SparkSession

#**********Begin**********#

#创建SparkSession
spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .config("spark.sql.crossJoin.enabled", "true") \
    .master("local") \
    .getOrCreate()
    
#读取/root/jun.json中数据
df =spark.read.json("/root/jun.json").coalesce(1)
#创建视图
df.createOrReplaceTempView("table1")

#统计出全球各研发单位研制的战斗机在全球所有战斗机中的占比
sqlDF = spark.sql("select concat(cast(round(count(`研发单位`)*100/(select count(`研发单位`) from table1 where `研发单位` is not null and `名称` is not null ),2) as float),'%'),`研发单位` from table1 where `研发单位` is not null and `名称` is not null group by `研发单位`")

#保存结果
sqlDF.write.format("csv").save("/root/airspark")
#**********End**********#

spark.stop()

第四章 Spark结构化流处理

军用大数据--结构化流式数据处理

第1关：Spark结构化流快速入门

任务描述
Spark Streaming 是一套优秀的实时计算框架。其良好的可扩展性、高吞吐量以及容错机制能够满足我们很多的场景应用。本关结合我们的应用场景，介结我们如何使用 Spark Streaming 处理数据。

# -*- coding: utf-8
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split
import time
# 请在此处编写代码
#********** Begin **********#
spark = SparkSession.builder.appName("StructuredNetworkWordCount").getOrCreate()
spark.sparkContext.setLogLevel("error")
lines = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load()
# Split the lines into words
words = lines.select(
   explode(
       split(lines.value, " ")
   ).alias("word")
)
# Generate running word count
wordCounts = words.groupBy("word").count()
 # Start running the query that prints the running counts to the console
query = wordCounts.writeStream.outputMode("complete").format("console").trigger(processingTime='1 seconds').start()
time.sleep( 20 )
query.stop()
#********** End **********#

第2关：对飞机的点击次数实时统计

任务描述
Spark Streaming 是一套优秀的实时计算框架。其良好的可扩展性、高吞吐量以及容错机制能够满足我们很多的场景应用。上一关我们介绍了如何实时获取数据，并处理数据，本关结合上一关的场景，结合 Kafka 把分析结果读出来。

相关知识
Spark Streaming 其优秀的特点给我们带来很多的应用场景。本关中，将通过从 TCP 获取数据来进行介绍。

import time
from pyspark.sql import SparkSession
#********** Begin **********#
spark = SparkSession.builder.master("local[2]").appName("case2").getOrCreate()
spark.sparkContext.setLogLevel("error")
df = spark.readStream.format("socket").option("host", "localhost").option("port", 9998).load()
ds=df.selectExpr( "CAST(value AS STRING)")
ds.createOrReplaceTempView("planeNumber")
sql= spark.sql("select count(*) nums,value from planeNumber   group  by value order by nums desc");
query =  sql.writeStream.format("console").outputMode("complete").start()
time.sleep( 20 )
query.stop()
#********** End **********#

kafka-入门篇

第1关：kafka - 初体验

任务描述
本关任务：使用 Kafka 命令创建一个副本数量为1、分区数量为3的 Topic 。

相关知识
为了完成本关任务，你需要掌握：1.如何使用 Kafka 的常用命令。

#!/bin/bash
#1.创建一个名为demo的Topic
kafka-topics.sh -create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 3 --topic demo
#2.查看所有Topic
kafka-topics.sh --list --zookeeper 127.0.0.1:2181
#3.查看名为demo的Topic的详情信息
kafka-topics.sh -topic demo --describe --zookeeper 127.0.0.1:2181

第2关：生产者 （Producer ） - 简单模式

任务描述
本关任务：编写一个 Kafka 的 Producer 进行数据生产。

相关知识
为了完成本关任务，你需要掌握：1.如何使用 Kafka 的 Producer API 进行数据生产。

package net.educoder;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**
 * kafka producer 简单模式
 */
public class App {
    public static void main(String[] args) {
        /**
         * 1.创建配置文件对象，一般采用 Properties
         */
        /**----------------begin-----------------------*/
        Properties props = new Properties();
        /**-----------------end-------------------------*/
        /**
         * 2.设置kafka的一些参数
         *          bootstrap.servers --> kafka的连接地址 127.0.0.1:9092
         *          key、value的序列化类 -->org.apache.kafka.common.serialization.StringSerializer
         *          acks：1，-1,0
         */
        /**-----------------begin-----------------------*/
        props.put("bootstrap.servers", "127.0.0.1:9092");
        props.put("acks", "1");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        /**-----------------end-------------------------*/
        /**
         * 3.构建kafkaProducer对象
         */
        /**-----------------begin-----------------------*/
        Producer<String, String> producer = new KafkaProducer<>(props);
        /**-----------------end-------------------------*/
        for (int i = 0; i < 100; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("demo", i + "", i + "");
            /**
             * 4.发送消息
             */
            /**-----------------begin-----------------------*/
            producer.send(record);
            /**-----------------end-------------------------*/
        }
        producer.close();
    }
}

第3关：消费者（ Consumer）- 自动提交偏移量

任务描述
本关任务：编写一个 Kafka 消费者并设置自动提交偏移量进行数据消费。

相关知识
为了完成本关任务，你需要掌握：1.如何编写 Kafka 消费者，2.如何使用自动提交偏移量。

package net.educoder;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
public class App {
    public static void main(String[] args) {
        Properties props = new Properties();
        /**--------------begin----------------*/
        //1.设置kafka集群的地址
        props.put("bootstrap.servers", "127.0.0.1:9092");
        //2.设置消费者组，组名字自定义，组名字相同的消费者在一个组
        props.put("group.id", "g1");
        //3.开启offset自动提交
        props.put("enable.auto.commit", "true");
        //4.自动提交时间间隔
        props.put("auto.commit.interval.ms", "1000");
        //5.序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        /**---------------end---------------*/
        /**--------------begin----------------*/
        //6.创建kafka消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //7.订阅kafka的topic
        consumer.subscribe(Arrays.asList("demo"));
        /**---------------end---------------*/
        int i = 1;
        while (true) {
            /**----------------------begin--------------------------------*/
            //8.poll消息数据,返回的变量为crs
            ConsumerRecords<String, String> crs = consumer.poll(100);
            for (ConsumerRecord<String, String> cr : crs) {
                System.out.println("consume data:" + i);
                i++;
            }
            /**----------------------end--------------------------------*/
            if (i > 10) {
                return;
            }
        }
    }
}

第4关：消费者（ Consumer ）- 手动提交偏移量

任务描述
本关任务：编写一个 Kafka 消费者并使用手动提交偏移量进行数据消费。

相关知识
为了完成本关任务，你需要掌握：1.如何编写 Kafka 消费者，2.如何手动提交偏移量。

Kafka 两种手动提交方式
异步提交( CommitAsync )：
   异步模式下，提交失败也不会尝试提交。消费者线程不会被阻塞，因为异步操作，可能在提交偏移量操作结果未返回时就开始下一次拉取操作。

同步提交( CommitSync )：
   同步模式下，提交失败时一直尝试提交，直到遇到无法重试才结束。同步方式下，消费者线程在拉取消息时会被阻塞，直到偏移量提交操作成功或者在提交过程中发生错误。

注意：实现手动提交前需要在创建消费者时关闭自动提交，设置enable.auto.commit=false

package net.educoder;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
public class App {
    public static void main(String[] args){
        Properties props = new Properties();
        /**-----------------begin------------------------*/
        //1.设置kafka集群的地址
        props.put("bootstrap.servers", "127.0.0.1:9092");
        //2.设置消费者组，组名字自定义，组名字相同的消费者在一个组
        props.put("group.id", "g1");
        //3.关闭offset自动提交
        props.put("enable.auto.commit", "false");
        //4.序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        /**-----------------end------------------------*/
        /**-----------------begin------------------------*/
        //5.实例化一个消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //6.消费者订阅主题，订阅名为demo的主题
        consumer.subscribe(Arrays.asList("demo"));
        /**-----------------end------------------------*/
        final int minBatchSize = 10;
        List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                buffer.add(record);
            }
            if (buffer.size() >= minBatchSize) {
                for (ConsumerRecord bf : buffer) {
                    System.out.printf("offset = %d, key = %s, value = %s%n", bf.offset(), bf.key(), bf.value());
                }
                /**-----------------begin------------------------*/
                //7.手动提交偏移量
                consumer.commitSync();
                /**-----------------end------------------------*/
                buffer.clear();
                return;
            }
        }
    }
}

第六章 Spark MLib机器学习

军用大数据 - Spark机器学习

第1关：Iris 分类

任务描述
本关任务：使用 pyspark ml 的LogisticRegression分类器完成 Iris 分类任务。

相关知识
观察数据集
我们本次使用的数据集是sklearn自带的数据集Iris。

接下来，我们来了解下Iris数据集的数据吧！

# -*- coding: utf-8 -*-
from pyspark.sql import SparkSession
from sklearn.datasets import load_iris
import pandas
from pyspark.ml.classification import LogisticRegression
from pyspark.mllib.evaluation import BinaryClassificationMetrics
from pyspark.ml.feature import RFormula


# 训练模型
def trainingModel(spark):
    # ********** Begin ********** #
    
    # 1.加载sklearn的训练数据
    iris =