ToB企服应用市场:ToB评测及商务社交产业平台

标题: 头歌实践教学平台大数据编程实训答案（三） [打印本页]

作者: 千千梦丶琪 时间: 2024-11-26 06:49
标题: 头歌实践教学平台大数据编程实训答案（三）
第一章遍历日记数据

用 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 =

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)