大数据处理技术-头歌平台-答案

这篇具有很好参考价值的文章主要介绍了大数据处理技术-头歌平台-答案。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

写在最前

这里是大数据处理技术的实训作业 ,学校使用的是“头歌”平台。(我已经不想吐槽了)
开始的几章很简单,所以没有写
其中有几章题目,仅仅需要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_111

 vim /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/
echo 1 > /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文件

echo 2 > /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文件

echo 3 > /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 代码

package com.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.
 */
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.apache.flume.interceptor.RegexExtractorInterceptorPassThroughSerializer;
import org.apache.flume.interceptor.RegexExtractorInterceptorSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;

public class RegexExtractorExtInterceptor implements Interceptor {  
  
    static final String REGEX = "regex";  
    static final String SERIALIZERS = "serializers";  
  
    // 增加代码开始  
  
    static final String EXTRACTOR_HEADER = "extractorHeader";  
    static final boolean DEFAULT_EXTRACTOR_HEADER = false;  
    static final String EXTRACTOR_HEADER_KEY = "extractorHeaderKey";  
  
    // 增加代码结束  
  
    private static final Logger logger = LoggerFactory  
            .getLogger(RegexExtractorExtInterceptor.class);  
  
    private final Pattern regex;  
    private final List<NameAndSerializer> serializers;  
  
    // 增加代码开始  
  
    private final boolean extractorHeader;  
    private final String extractorHeaderKey;  
  
    // 增加代码结束  
  
    private RegexExtractorExtInterceptor(Pattern regex,  
            List<NameAndSerializer> serializers, boolean extractorHeader,  
            String extractorHeaderKey) {  
        this.regex = regex;  
        this.serializers = serializers;  
        this.extractorHeader = extractorHeader;  
        this.extractorHeaderKey = extractorHeaderKey;  
    }  
  
    @Override  
    public void initialize() {  
        // NO-OP...  
    }  
  
    @Override  
    public void close() {  
        // NO-OP...  
    }  
  
    @Override  
    public Event intercept(Event event) {  
        String tmpStr;  
        if(extractorHeader)  
        {  
            tmpStr = event.getHeaders().get(extractorHeaderKey);  
        }  
        else  
        {  
            tmpStr=new String(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;  
    }  
  
    @Override  
    public List<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;  
    }  
  
    public static class Builder implements Interceptor.Builder {  
  
        private Pattern regex;  
        private List<NameAndSerializer> serializerList;  
  
        // 增加代码开始  
  
        private boolean extractorHeader;  
        private String extractorHeaderKey;  
  
        // 增加代码结束  
  
        private final RegexExtractorInterceptorSerializer defaultSerializer = new RegexExtractorInterceptorPassThroughSerializer();  
  
        @Override  
        public void configure(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");  
            }  
            // 增加代码结束  
        }  
  
        private void configureSerializers(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 = new Context(  
                    context.getSubProperties(SERIALIZERS + "."));  
  
            serializerList = Lists  
                    .newArrayListWithCapacity(serializerNames.length);  
            for (String serializerName : serializerNames) {  
                Context serializerContext = new Context(  
                        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(new NameAndSerializer(name,  
                            defaultSerializer));  
                } else {  
                    serializerList.add(new NameAndSerializer(name,  
                            getCustomSerializer(type, serializerContext)));  
                }  
            }  
        }  
  
        private RegexExtractorInterceptorSerializer getCustomSerializer(  
                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;  
        }  
  
        @Override  
        public Interceptor build() {  
            Preconditions.checkArgument(regex != null,  
                    "Regex pattern was misconfigured");  
            Preconditions.checkArgument(serializerList.size() > 0,  
                    "Must supply a valid group match id list");  
            return new RegexExtractorExtInterceptor(regex, serializerList,  
                    extractorHeader, extractorHeaderKey);  
        }  
    }  
  
    static class NameAndSerializer {  
        private final String headerName;  
        private final RegexExtractorInterceptorSerializer serializer;  
  
        public NameAndSerializer(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的错误。**系统。
多试几次

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.创建配置文件对象,一般采用 props
         */

        /**----------------begin-----------------------*/
 Properties props = new Properties();
        

        /**-----------------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 = new KafkaProducer<String, String>(props);

        /**-----------------end-------------------------*/

        for (int i = 0; i < 2; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("demo", ""+i);
            /**
             * 4.发送消息
             */

            /**-----------------begin-----------------------*/

            producer.send(record);

            /**-----------------end-------------------------*/
        }
        producer.close();
    }
}


第3关:消费者( Consumer)- 自动提交偏移量

有时候会报scala的错误。**系统。
多试几次

然后也要吐槽一下示例的代码, 少个" 是什么鬼。
而且也没有缺少提示, 真应该抓他们过来,让他们一个个给我找!

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----------------*/

        //设置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 = 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 )- 手动提交偏移量
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");
//设置消费者组,组名字自定义,组名字相同的消费者在一个组
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 = 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 Standalone 模式的安装和部署

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

吐槽:
这一关的任务要求部分给的一点都不好。
一点不人性化, 其他的不吐槽了。 “小白”要在这个平台做这道题,恶心不死你。文章来源地址https://www.toymoban.com/news/detail-425509.html

到了这里,关于大数据处理技术-头歌平台-答案的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 头歌平台-人工智能技术应用-实践学习与答案2(补充实训部分)

    注:这一题的输出没有很符合我的预期,所以我干脆直接改了他的print输出,用自己更喜欢的方式输出 注: 这里对字典的统计我引入了defaultdict函数(这个函数是用来新建一个键值对的),算是额外引入了一个算法库使用 测试用例: 一、 针对集体宿舍人员如何科学防控的问

    2024年02月07日
    浏览(73)
  • 头歌实践教学平台数据库原理与应用实训答案

    目录 实训一:数据定义和操纵(4课时) 初识MySQL数据库 第1关:创建数据库  第2关:创建表  第3关:使用主键约束 第4关:外键约束 第5关:添加常用约束 DDL语言的使用 第1关:创建数据库  第2关: 创建表  第3关:添加字段  第4关:删除字段  第5关:修改字段  第6关:添加

    2024年02月08日
    浏览(49)
  • 云计算与大数据第9章 大数据处理习题带答案

    1、在数据预处理阶段,数据合并到一致的存储介质中,使得数据挖掘更有效、挖掘模式更易理解,这一过程是(  B  )。 A. 数据清洗         B. 数据集成        C. 数据归约      D.数据转换    2、以下(  B  )不是数据归约策略。 A. 属性子集的选择   B. 属性构

    2024年02月09日
    浏览(46)
  • 深入剖析:Kafka流数据处理引擎的核心面试问题解析75问(5.7万字参考答案)

           Kafka 是一款开源的分布式流处理平台,被广泛应用于构建实时数据管道、日志聚合、事件驱动的架构等场景。本文将深入探究 Kafka 的基本原理、特点以及其在实际应用中的价值和作用。      Kafka 的基本原理是建立在发布-订阅模式之上的。生产者将消息发布到主题(

    2024年01月23日
    浏览(47)
  • 【头歌】——数据分析与实践-基于Python语言的文件与文件夹管理-文本 文件处理-利用csv模块进行csv文件的读写操作

    第1关 创建子文件夹 第2关 删除带有只读属性的文件 第3关 批量复制文件夹中的所有文件 未通过本题,如果您通过了本题欢迎补充到评论区,有时间我会整理进来 第1关 读取宋词文件,根据词人建立多个文件 第2关 读取宋词文件,并根据词人建立多个文件夹 第3关 读取宋词文

    2024年01月25日
    浏览(73)
  • 大数据处理平台的架构演进:从批处理到实时流处理

    🎈个人主页:程序员 小侯 🎐CSDN新晋作者 🎉欢迎 👍点赞✍评论⭐收藏 ✨收录专栏:大数据系列 ✨文章内容:大数据框架演进 🤝希望作者的文章能对你有所帮助,有不足的地方请在评论区留言指正,大家一起学习交流!🤗 大数据处理平台的架构演进经历了从批处理到实

    2024年02月10日
    浏览(49)
  • Pandas初体验——头歌平台答案

    第1关 了解数据处理对象--Series 第2关 了解数据处理对象-DataFrame 第3关 读取CSV格式数据 第4关 数据的基本操作——排序 第5关 数据的基本操作——删除 第6关 数据的基本操作——算术运算 第7关 数据的基本操作——去重 第8关 层次化索引 如果显示评测未通过,不要怀疑自己,

    2024年02月11日
    浏览(36)
  • 头歌实践教学平台答案(Java实训作业答案)

    搜集整理了一份最新最全的头歌(EduCoder)Java实训作业答案,分享给大家.(EduCoder)是信息技术类实践教学平台。(EduCoder)涵盖了计算机、大数据、云计算、人工智能、软件工程、物联网等专业课程。超60000个实训案例,建立学、练、评、测一体化实验环境。这份是头歌实践教学平

    2023年04月11日
    浏览(84)
  • 头歌实践教学平台答案(消息队列面试题及答案)

    头歌实践教学平台消息队列 面试题及答案,消息队列⾯试题及答案。 1、为什么使用消息队列? 消息队列使用的场景和中间件有很多,但解决的核心问题主要是:异步、解耦、消峰填谷。 异步、解耦、消峰填谷这是消息队列最大的优点,除了这些消息队列还可以会解决一些

    2024年02月11日
    浏览(68)
  • 使用AI平台处理训练和微调数据

                  Llama.cpp是Georgi Gerganov 基于 Meta 的 LLaMA 模型 手写的纯 C/C++ 版本,让我们实现了在笔记本电脑上部署和体验AI大模型,实现没有GPU也可以运行AI大模型。执行起来虽然比较慢,但是只能算做体验,还可以选择不同语言。某个模型使用体验不好时,还可以更换

    2024年01月19日
    浏览(37)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包