文章目录
写在最前
这里是大数据处理技术的实训作业 ，学校使用的是“头歌”平台。（我已经不想吐槽了）
开始的几章很简单，所以没有写
其中有几章题目，仅仅需要ctrl+c ctrl+v即可，只是操作步骤麻烦一下，所以也没有写。
HBase的安装与简单操作
第一关：单机版安装
mkdir /app
cd /opt
tar -zxvf hbase-2.1.1-bin.tar.gz -C /app
vim /app/hbase-2.1.1/conf/hbase-env.sh
# 在末尾添加 export JAVA_HOME=/usr/lib/jvm/jdk1.8.0_111vim /app/hbase-2.1.1/conf/hbase-site.xml
替换原有的configuration标签
<configuration><property><name>hbase.rootdir</name><value>file:///root/data/hbase/data</value></property><property><name>hbase.zookeeper.property.dataDir</name><value>/root/data/hbase/zookeeper</value></property><property><name>hbase.unsafe.stream.capability.enforce</name><value>false</value></property></configuration>
vim /etc/profile
# 在末尾追加如下内容
#SET HBASE_enviroment
HBASE_HOME=/app/hbase-2.1.1
export PATH=$PATH:$HBASE_HOME/bin
source /etc/profile
第三关
put 'mytable','row1','data:1','zhangsan'
put 'mytable','row2','data:2','zhangsanfeng'
put 'mytable','row3','data:3','zhangwuji'
HBase 伪分布式环境搭建
第一关：伪分布式环境搭建
先按照 《HBase的安装与简单第一关配置好单机》，傻子平台。
vim /app/hbase-2.1.1/conf/hbase-site.xml
<!-- 替换configuration整体 --><configuration><property><name>hbase.rootdir</name><value>hdfs://localhost:9000/hbase</value></property><property><name>hbase.zookeeper.property.dataDir</name><value>/root/data/hbase/zookeeper</value></property><property><name>hbase.unsafe.stream.capability.enforce</name><value>true</value></property><property><name>hbase.cluster.distributed</name><value>true</value></property></configuration>
# 启动hadoop和hbase
start-all.sh
start-hbase.sh
# 查看进程
jps
# 在hdfs中验证
hadoop fs -ls /hbase
ZooKeeper入门-初体验
第一关 ZooKeeper初体验
tar -zxvf zookeepre-3.4.12.tar.gz /opt/zookeeper-3.4.12
cd /opt/zookeeper-3.4.12/conf
mv zoo_sample.cfg zoo.cfg
zkServer.sh start
# zkServer.sh stop
第2关：ZooKeeper配置
vim /opt/zookeeper-3.4.12/conf/zoo.cfg
把 “# maxClientCnxns=60 ”
改为
maxClientCnxns=100
第3关：Client连接及状态
zkServer.sh stop
vim /opt/zookeeper-3.4.12/conf/zoo.cfg
<!-- 修改为2182 -->
clientPort=2182
<!-- 添加preAllocSize=300 -->
preAllocSize=300
vim /opt/zookeeper-3.4.12/bin/zkEnv.sh
<!-- 修改第56行为 -->
ZOO_LOG_DIR="/opt/zookeeper-3.4.12"
zkServer.sh start
zkCli.sh -server 127.0.0.1:2182
ZooKeeper之分布式环境搭建
第1关：仲裁模式与伪分布式环境搭建
vim /opt/zookeeper-3.4.12/conf/zoo.cfg
修改默认。 修改zoo.cfg
这节有个智障操作，这里不吐槽了。按着步骤走吧。
<!-- zookeeper-3.4.12的zoo.cfg --><!-- 修改 -->
clientPort=2181
dataDir=/opt/zookeeper-3.4.12/tmp/data
<!-- 末尾追加 -->
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890
第一个节点添加myid文件
mkdir -p /opt/zookeeper-3.4.12/tmp/data/
echo1> /opt/zookeeper-3.4.12/tmp/data/myid
cat /opt/zookeeper-3.4.12/tmp/data/myid
复制三个新节点出来
# 智障系统。您搁着我斗志斗勇呢呀cp -r /opt/zookeeper-3.4.12/ /opt/zookeeper-3.4.12-01
cp -r /opt/zookeeper-3.4.12/ /opt/zookeeper-3.4.12-02
cp -r /opt/zookeeper-3.4.12/ /opt/zookeeper-3.4.12-03
第一个节点 修改zoo.cfg
vim /opt/zookeeper-3.4.12-01/conf/zoo.cfg
<!-- zookeeper-3.4.12-01的zoo.cfg --><!-- 仅修改这个就行 -->
dataDir=/opt/zookeeper-3.4.12-01/tmp/data
第二个节点 修改zoo.cfg
vim /opt/zookeeper-3.4.12-02/conf/zoo.cfg
<!-- zookeeper-3.4.12-02的zoo.cfg --><!-- 修改 -->
clientPort=2182
dataDir=/opt/zookeeper-3.4.12-02/tmp/data
第二个节点添加myid文件
echo2> /opt/zookeeper-3.4.12-02/tmp/data/myid
cat /opt/zookeeper-3.4.12-02/tmp/data/myid
第三个节点 修改zoo.cfg
vim /opt/zookeeper-3.4.12-03/conf/zoo.cfg
<!-- zookeeper-3.4.12-03的zoo.cfg --><!-- 修改 -->
clientPort=2183
dataDir=/opt/zookeeper-3.4.12-03/tmp/data
第三个节点添加myid文件
echo3> /opt/zookeeper-3.4.12-03/tmp/data/myid
cat /opt/zookeeper-3.4.12-03/tmp/data/myid
# 分别三个启动节点
/opt/zookeeper-3.4.12-01/bin/zkServer.sh start
/opt/zookeeper-3.4.12-02/bin/zkServer.sh start
/opt/zookeeper-3.4.12-03/bin/zkServer.sh start
第2关：伪分布式体验及分布式安装配置
智障平台，我重置了一次命令行，重新做了一遍才行。
zkCli.sh -server 127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
create /quorum_test "quorum_test"
quit
Flume入门
第1关：Flume 简介
第一题
Source Channel Sink
第二题
名称 类型 属性集
第三题
可靠性 可恢复性
第2关：采集目录下所有新文件到Hdfs
start-dfs.sh
hadoop dfs -mkdir /flume
我不得不吐槽一下这个平台。
你说你资源不够你做什么平台嘛。
也是，我理解，随时启动一个hadoop确实很耗费资源，但你不能在启动脚本中再启动一次hadoop吗？ 你在这跟我捉迷藏呢？真就担心我找到你哈？
a1.sources = source1
a1.sinks = sink1
a1.channels = channel1
# 配置source组件
a1.sources.source1.type = spooldir
a1.sources.source1.spoolDir = /opt/flume/data
##定义文件上传完后的后缀,默认是.COMPLETED
a1.sources.source1.fileSuffix=.FINISHED
##默认是2048,如果文件行数据量超过2048字节(1k),会被截断，导致数据丢失
a1.sources.source1.deserializer.maxLineLength=5120
# 配置sink组件
a1.sinks.sink1.type = hdfs
a1.sinks.sink1.hdfs.path =hdfs://localhost:9000/flume
#上传文件的前缀
a1.sinks.sink1.hdfs.filePrefix = flume
#上传文件的后缀
a1.sinks.sink1.hdfs.fileSuffix = .log
#积攒多少个Event才flush到HDFS一次
a1.sinks.sink1.hdfs.batchSize= 100
a1.sinks.sink1.hdfs.fileType = DataStream
a1.sinks.sink1.hdfs.writeFormat =Text
## roll：滚动切换：控制写文件的切换规则
## 按文件体积（字节）来切
a1.sinks.sink1.hdfs.rollSize = 512000
## 按event条数切
a1.sinks.sink1.hdfs.rollCount = 1000000
## 按时间间隔切换文件，多久生成一个新的文件
a1.sinks.sink1.hdfs.rollInterval = 4
## 控制生成目录的规则
a1.sinks.sink1.hdfs.round = true
##多少时间单位创建一个新的文件夹
a1.sinks.sink1.hdfs.roundValue = 10
a1.sinks.sink1.hdfs.roundUnit = minute
#是否使用本地时间戳
a1.sinks.sink1.hdfs.useLocalTimeStamp = true
# channel组件配置
a1.channels.channel1.type = memory
## event条数
a1.channels.channel1.capacity = 500000
##flume事务控制所需要的缓存容量600条event
a1.channels.channel1.transactionCapacity = 600
# 绑定source、channel和sink之间的连接
a1.sources.source1.channels = channel1
a1.sinks.sink1.channel = channel1
Flume进阶
第1关：拦截器的使用
start-dfs.sh
hadoop dfs -mkdir /flume
# Define source, channel, sink
#agent名称为a1
# Define source
#source类型配置为avro,监听8888端口，后台会自动发送数据到该端口
#拦截后台发送过来的数据，将y.开头的保留下来
# Define channel
#channel配置为memery
# Define sink
#落地到 hdfs://localhost:9000/flume目录下
#根据时间落地，3s
#数据格式DataStream
a1.sources = source1
a1.sinks = sink1
a1.channels = channel1
# 配置source组件
a1.sources.source1.type = avro
a1.sources.source1.bind = 127.0.0.1
a1.sources.source1.port = 8888
##定义文件上传完后的后缀,默认是.COMPLETED
a1.sources.source1.fileSuffix=.FINISHED
##默认是2048,如果文件行数据量超过2048字节(1k),会被截断，导致数据丢失
a1.sources.source1.deserializer.maxLineLength=5120
#正则过滤拦截器
a1.sources.source1.interceptors = i1
a1.sources.source1.interceptors.i1.type = regex_filter
a1.sources.source1.interceptors.i1.regex = ^y.*
#如果excludeEvents设为false,表示过滤掉不是以A开头的events。
#如果excludeEvents设为true，则表示过滤掉以A开头的events。
a1.sources.source1.interceptors.i1.excludeEvents = false
# 配置sink组件
a1.sinks.sink1.type = hdfs
a1.sinks.sink1.hdfs.path =hdfs://localhost:9000/flume
#上传文件的前缀
a1.sinks.sink1.hdfs.filePrefix = FlumeData.
#上传文件的后缀
a1.sinks.sink1.hdfs.fileSuffix = .log
#积攒多少个Event才flush到HDFS一次
a1.sinks.sink1.hdfs.batchSize= 100
a1.sinks.sink1.hdfs.fileType = DataStream
a1.sinks.sink1.hdfs.writeFormat =Text
## roll：滚动切换：控制写文件的切换规则
## 按文件体积（字节）来切
a1.sinks.sink1.hdfs.rollSize = 512000
## 按event条数切
a1.sinks.sink1.hdfs.rollCount = 1000000
## 按时间间隔切换文件，多久生成一个新的文件
a1.sinks.sink1.hdfs.rollInterval = 4
## 控制生成目录的规则
a1.sinks.sink1.hdfs.round = true
##多少时间单位创建一个新的文件夹
a1.sinks.sink1.hdfs.roundValue = 10
a1.sinks.sink1.hdfs.roundUnit = minute
#是否使用本地时间戳
a1.sinks.sink1.hdfs.useLocalTimeStamp = true
# channel组件配置
a1.channels.channel1.type = memory
## event条数
a1.channels.channel1.capacity = 500000
##flume事务控制所需要的缓存容量600条event
a1.channels.channel1.transactionCapacity = 600
# 绑定source、channel和sink之间的连接
a1.sources.source1.channels = channel1
a1.sinks.sink1.channel = channel1
第2关：自定义拦截器
参考链接
conf 配置文件
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# The configuration file needs to define the sources,
# the channels and the sinks.
# Sources, channels and sinks are defined per agent,
# in this case called 'agent'
# Define source, channel, sink
#agent名为a1;
# Define and configure an Spool directory source
#采集 /opt/flume/data目录下所有文件
# Configure channel
#channel选择memery
# Define and configure a hdfs sink
#落地到hdfs的hdfs://localhost:9000/flume/文件名的前缀/文件名上的日期
#文件格式设为DataStream
#根据时间回滚，3s
a1.sources=source1
a1.channels=channel1
a1.sinks=sink1
a1.sources.source1.type=spooldir
a1.sources.source1.spoolDir=/opt/flume/data
a1.sources.source1.fileHeader=true
a1.sources.source1.basenameHeader=true
a1.sources.source1.interceptors=i1
a1.sources.source1.interceptors.i1.type=com.yy.RegexExtractorExtInterceptor$Builder
a1.sources.source1.interceptors.i1.regex=(.*)\\.(.*)\\.(.*)
a1.sources.source1.interceptors.i1.extractorHeader=true
a1.sources.source1.interceptors.i1.extractorHeaderKey=basename
a1.sources.source1.interceptors.i1.serializers=s1 s2 s3
a1.sources.source1.interceptors.i1.serializers.s1.name=one
a1.sources.source1.interceptors.i1.serializers.s2.name=two
a1.sources.source1.interceptors.i1.serializers.s3.name=three
a1.sources.source1.channels=channel1
a1.sinks.sink1.type=hdfs
a1.sinks.sink1.channel=channel1
a1.sinks.sink1.hdfs.path=hdfs://localhost:9000/flume/%{one}/%{three}
a1.sinks.sink1.hdfs.round=true
a1.sinks.sink1.hdfs.roundValue=10
a1.sinks.sink1.hdfs.roundUnit=minute
a1.sinks.sink1.hdfs.fileType=DataStream
a1.sinks.sink1.hdfs.writeFormat=Text
a1.sinks.sink1.hdfs.rollInterval=0
a1.sinks.sink1.hdfs.rollSize=10240
a1.sinks.sink1.hdfs.rollCount=0
a1.sinks.sink1.hdfs.idleTimeout=60
a1.channels.channel1.type=memory
a1.channels.channel1.capacity=10000
a1.channels.channel1.transactionCapacity=1000
a1.channels.channel1.keep-alive=30
java 代码
packagecom.yy;/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/importjava.util.List;importjava.util.Map;importjava.util.regex.Matcher;importjava.util.regex.Pattern;importorg.apache.commons.lang.StringUtils;importorg.apache.flume.Context;importorg.apache.flume.Event;importorg.apache.flume.interceptor.Interceptor;importorg.apache.flume.interceptor.RegexExtractorInterceptorPassThroughSerializer;importorg.apache.flume.interceptor.RegexExtractorInterceptorSerializer;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importcom.google.common.base.Charsets;importcom.google.common.base.Preconditions;importcom.google.common.base.Throwables;importcom.google.common.collect.Lists;publicclassRegexExtractorExtInterceptorimplementsInterceptor{staticfinalString REGEX ="regex";staticfinalString SERIALIZERS ="serializers";// 增加代码开始 staticfinalString EXTRACTOR_HEADER ="extractorHeader";staticfinalboolean DEFAULT_EXTRACTOR_HEADER =false;staticfinalString EXTRACTOR_HEADER_KEY ="extractorHeaderKey";// 增加代码结束 privatestaticfinalLogger logger =LoggerFactory.getLogger(RegexExtractorExtInterceptor.class);privatefinalPattern regex;privatefinalList<NameAndSerializer> serializers;// 增加代码开始 privatefinalboolean extractorHeader;privatefinalString extractorHeaderKey;// 增加代码结束 privateRegexExtractorExtInterceptor(Pattern regex,List<NameAndSerializer> serializers,boolean extractorHeader,String extractorHeaderKey){this.regex = regex;this.serializers = serializers;this.extractorHeader = extractorHeader;this.extractorHeaderKey = extractorHeaderKey;}@Overridepublicvoidinitialize(){// NO-OP... }@Overridepublicvoidclose(){// NO-OP... }@OverridepublicEventintercept(Event event){String tmpStr;if(extractorHeader){
tmpStr = event.getHeaders().get(extractorHeaderKey);}else{
tmpStr=newString(event.getBody(),Charsets.UTF_8);}Matcher matcher = regex.matcher(tmpStr);Map<String,String> headers = event.getHeaders();if(matcher.find()){for(int group =0, count = matcher.groupCount(); group < count; group++){int groupIndex = group +1;if(groupIndex > serializers.size()){if(logger.isDebugEnabled()){
logger.debug("Skipping group {} to {} due to missing serializer",
group, count);}break;}NameAndSerializer serializer = serializers.get(group);if(logger.isDebugEnabled()){
logger.debug("Serializing {} using {}",
serializer.headerName, serializer.serializer);}
headers.put(serializer.headerName, serializer.serializer
.serialize(matcher.group(groupIndex)));}}return event;}@OverridepublicList<Event>intercept(List<Event> events){List<Event> intercepted =Lists.newArrayListWithCapacity(events.size());for(Event event : events){Event interceptedEvent =intercept(event);if(interceptedEvent !=null){
intercepted.add(interceptedEvent);}}return intercepted;}publicstaticclassBuilderimplementsInterceptor.Builder{privatePattern regex;privateList<NameAndSerializer> serializerList;// 增加代码开始 privateboolean extractorHeader;privateString extractorHeaderKey;// 增加代码结束 privatefinalRegexExtractorInterceptorSerializer defaultSerializer =newRegexExtractorInterceptorPassThroughSerializer();@Overridepublicvoidconfigure(Context context){String regexString = context.getString(REGEX);Preconditions.checkArgument(!StringUtils.isEmpty(regexString),"Must supply a valid regex string");
regex =Pattern.compile(regexString);
regex.pattern();
regex.matcher("").groupCount();configureSerializers(context);// 增加代码开始
extractorHeader = context.getBoolean(EXTRACTOR_HEADER,
DEFAULT_EXTRACTOR_HEADER);if(extractorHeader){
extractorHeaderKey = context.getString(EXTRACTOR_HEADER_KEY);Preconditions.checkArgument(!StringUtils.isEmpty(extractorHeaderKey),"必须指定要抽取内容的header key");}// 增加代码结束 }privatevoidconfigureSerializers(Context context){String serializerListStr = context.getString(SERIALIZERS);Preconditions.checkArgument(!StringUtils.isEmpty(serializerListStr),"Must supply at least one name and serializer");String[] serializerNames = serializerListStr.split("\\s+");Context serializerContexts =newContext(
context.getSubProperties(SERIALIZERS +"."));
serializerList =Lists.newArrayListWithCapacity(serializerNames.length);for(String serializerName : serializerNames){Context serializerContext =newContext(
serializerContexts.getSubProperties(serializerName
+"."));String type = serializerContext.getString("type","DEFAULT");String name = serializerContext.getString("name");Preconditions.checkArgument(!StringUtils.isEmpty(name),"Supplied name cannot be empty.");if("DEFAULT".equals(type)){
serializerList.add(newNameAndSerializer(name,
defaultSerializer));}else{
serializerList.add(newNameAndSerializer(name,getCustomSerializer(type, serializerContext)));}}}privateRegexExtractorInterceptorSerializergetCustomSerializer(String clazzName,Context context){try{RegexExtractorInterceptorSerializer serializer =(RegexExtractorInterceptorSerializer)Class.forName(clazzName).newInstance();
serializer.configure(context);return serializer;}catch(Exception e){
logger.error("Could not instantiate event serializer.", e);Throwables.propagate(e);}return defaultSerializer;}@OverridepublicInterceptorbuild(){Preconditions.checkArgument(regex !=null,"Regex pattern was misconfigured");Preconditions.checkArgument(serializerList.size()>0,"Must supply a valid group match id list");returnnewRegexExtractorExtInterceptor(regex, serializerList,
extractorHeader, extractorHeaderKey);}}staticclassNameAndSerializer{privatefinalString headerName;privatefinalRegexExtractorInterceptorSerializer serializer;publicNameAndSerializer(String headerName,RegexExtractorInterceptorSerializer serializer){this.headerName = headerName;this.serializer = serializer;}}}
分布式 Kafka 安装
第1关：分布式 Kafka 安装
这关平台左侧给的示例中。有一条使用了中文的逗号。要自己改成英文的。这点注意⚠️
这里原本评判脚本有问题。
向工程师提交后，对方修改。
而后按照顺序走即可
kafka-入门篇
第1关：kafka - 初体验
#1.创建一个副本数量为1、分区数量为3、名为 demo 的 Topic
/opt/kafka_2.11-1.1.0/bin/kafka-topics.sh --create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 3 --topic demo
#2.查看所有Topic
/opt/kafka_2.11-1.1.0/bin/kafka-topics.sh --list --zookeeper 127.0.0.1:2181
#3.查看名为demo的Topic的详情信息
/opt/kafka_2.11-1.1.0/bin/kafka-topics.sh --topic demo --describe --zookeeper 127.0.0.1:2181
第2关：生产者 （Producer ） - 简单模式
有时候会报scala的错误。**系统。
多试几次
packagenet.educoder;importorg.apache.kafka.clients.producer.KafkaProducer;importorg.apache.kafka.clients.producer.Producer;importorg.apache.kafka.clients.producer.ProducerRecord;importjava.util.Properties;/**
* kafka producer 简单模式
*/publicclassApp{publicstaticvoidmain(String[] args){/**
* 1.创建配置文件对象，一般采用 props
*//**----------------begin-----------------------*/Properties props =newProperties();/**-----------------end-------------------------*//**
* 2.设置kafka的一些参数
* bootstrap.servers --> kafka的连接地址 kafka-01:9092,kafka-02:9092,kafka-03: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("bootstrap.servers", "kafka-01:9092,kafka-02:9092,kafka-03:9092");// props.put("bootstrap.servers","127.0.0.1:2181")
props.put("acks","1");
props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
props.put("retries",0);// 一批消息的处理大小
props.put("batch.size",16384);// 请求的延迟
props.put("linger.ms",1);// 发送缓冲区内存大小
props.put("buffer.size",33554432);// // key 序列化// props.put("key.serializer", "org.apache.kafka.common.serilization.StringSerilizer");// // value 序列化// props.put("value.serializer", "org.apache.kafka.common.serilization.StringSerilizer");// KafkaProducer 有多个构造方法，可以用Map来进行社会参数，也可在构造方法中进行设置序列化/**-----------------end-------------------------*//**
* 3.构建kafkaProducer对象
*//**-----------------begin-----------------------*/// Producer<String, String> producer = new KafkaProducer<>(props);KafkaProducer producer =newKafkaProducer<String,String>(props);/**-----------------end-------------------------*/for(int i =0; i <2; i++){ProducerRecord<String,String>record=newProducerRecord<>("demo",""+i);/**
* 4.发送消息
*//**-----------------begin-----------------------*/
producer.send(record);/**-----------------end-------------------------*/}
producer.close();}}
第3关：消费者（ Consumer）- 自动提交偏移量
有时候会报scala的错误。**系统。
多试几次
然后也要吐槽一下示例的代码， 少个" 是什么鬼。
而且也没有缺少提示， 真应该抓他们过来，让他们一个个给我找！
packagenet.educoder;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importjava.util.Arrays;importjava.util.Properties;publicclassApp{publicstaticvoidmain(String[] args){Properties props =newProperties();/**--------------begin----------------*///设置kafka集群的地址
props.put("bootstrap.servers","127.0.0.1:9092");//设置消费者组，组名字自定义，组名字相同的消费者在一个组
props.put("group.id","g1");//开启offset自动提交
props.put("enable.auto.commit","true");//自动提交时间间隔
props.put("auto.commit.interval.ms","1000");//序列化器
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 =newKafkaConsumer<>(props);//7.订阅kafka的topic
consumer.subscribe(Arrays.asList("demo"));/**---------------end---------------*/int i =1;while(true){/**----------------------begin--------------------------------*///8.poll消息数据,返回的变量为crsConsumerRecords<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 ）- 手动提交偏移量
packagenet.educoder;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importjava.util.ArrayList;importjava.util.Arrays;importjava.util.List;importjava.util.Properties;publicclassApp{publicstaticvoidmain(String[] args){Properties props =newProperties();/**-----------------begin------------------------*///1.设置kafka集群的地址
props.put("bootstrap.servers","127.0.0.1:9092");//设置消费者组，组名字自定义，组名字相同的消费者在一个组
props.put("group.id","g1");//3.关闭offset自动提交
props.put("enable.auto.commit","false");
props.put("max.poll.records",10);//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 =newKafkaConsumer<>(props);//6.消费者订阅主题，订阅名为demo的主题
consumer.subscribe(Arrays.asList("demo"));/**-----------------end------------------------*/finalint minBatchSize =10;List<ConsumerRecord<String,String>> buffer =newArrayList<>();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 Standalone 模式的安装和部署
第1关： Standalone 分布式集群搭建
吐槽:
这一关的任务要求部分给的一点都不好。
一点不人性化， 其他的不吐槽了。 “小白”要在这个平台做这道题，恶心不死你。
本文转载自: https://blog.csdn.net/qq_42182429/article/details/121751841
版权归原作者 从化北 所有， 如有侵权，请联系我们删除。
版权归原作者 从化北 所有， 如有侵权，请联系我们删除。