>> 欢迎您 客人: 登录 | 注册 | 在线 | 搜索 | 风格 | 帮助 | 统计 | 文章 返回首页
>>分享SPSS,Hadoop等大数据处理技术，以及分布式架构以及集群系统的构建 书籍支持  卫琴直播  品书摘要  在线测试  资源下载  联系我们
Java技术网站 → 『 Java开发技术 』 → 『 大数据处理和分布式架构专题 』 → 大数据项目之仿天猫数据分析
发表一个新主题 开启一个新投票 回复文章 您是本文章第 22371 个阅读者 刷新本主题
 * 贴子主题:  大数据项目之仿天猫数据分析 回复文章 点赞(0)  收藏  
作者:flybird    发表时间：2024-04-19 01:41:06     消息  查看  搜索  好友  邮件  复制  引用
    

简介

项目简介

此项目是实现仿大数据项目流程，包括，日志收集传输，日志格式化，数据实时分析，数据持久化到HDFS，数据离线报表统计，离线任务调度，日志记录搜索几大部分

首先说一下这个项目的大致流程，这个项目是仿天猫数据分析，是自己模拟用户购买浏览商品，生成日志，对日志进行收集，然后分两部分，一部分对数据实时分析，生成日用户活跃地理位置，第二部分是数据持久化之后，T+1对数据进行分析，统计各省销量以及各省活跃用户数

项目数据流

  1. 日志数据生成

    日志格式: requestid, ts, userid, 城市,经度,纬度,操作(浏览,购买)
  2. flume采集日志数据进入kafka log topic
  3. kafkastream消费log topic日志,写入process topic

    对数据进行格式化处理，以及过滤数据。格式化后的数据格式: requestid, ts, userid, 城市,经度,纬度,操作
  4. 实时模块

    sparkstreaming处理process topic日志，扔进realtime topic

    格式化数据，写到process topic中，得到城市和用户id
  5. flume采集process topic数据，写入hdfs

    持久化日志到hdfs中
  6. report模块(离线处理模块)

    T+1离线处理模块，spark计算hdfs中的数据,统计前一天的各省销售记录,写入mysql
  7. azkaban调度任务
  8. ElasticSearch查询历史记录
  9. web页面实时展示活跃用户，和报表页面
然而。。电脑垃圾，扛不住，只做了一部分

azkaban，elasticsearch，web页面没有做

项目构成

  1. logbuilder
模拟日志生成(后面为了方便，写了一个shell用于日志生成)

  1. kafkastream

    kafkastream清洗日志
  2. sparkstream

    sparkstream实时处理日志显示操作用户的地理位置
  3. report

    T+1报表项目,批处理日志，分析各省销售量对比,写入mysql

实现

我们按照数据流的方式来写实现

1. 日志生成

使用SpringBoot做了一个日志生成器。模拟生成日志，日志格式


logger:>>>> requestid,userid,ts,城市，经度，维度，操作(0浏览 1购买)

public static void main(String[] args) {

        SpringApplication.run(LogBuilderApplication.class);

        HashMap map = new HashMap<Integer, String[]>();

        // 存储城市，经纬度
        map.put(0,new String[]{"海门","121.15","31.89"});
        map.put(1,new String[]{"盐城","120.13","33.38"});
        map.put(2,new String[]{"上海","121.48","31.22"});
        map.put(3,new String[]{"厦门","118.1","24.46"});
        Random random = new Random();

        while (true){
            int i1 = random.nextInt(map.size());
            String[] o = (String[]) map.get(i1);
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            long ts = System.currentTimeMillis();
            String requestid = UUID.randomUUID().toString();

            int userid = random.nextInt(100000);
            // requestid,userid, 城市,经度,纬度,操作(浏览,购买)0 浏览 1 购买
            logger.info("logger:>>>>{},{},{},{},{},{},{}",requestid, ts, userid, o[0],o[1],o[2],random.nextInt(2));
        }
    }

  由于不想打包，重启，就自己用shell写了一个日志生成器，更好使用


#!/bin/bash
#二维数组
city=('海门 121.15 31.89' '盐城 120.13 33.38' '上海 121.48 31.22' '厦门 118.1 24.46')
#city=('海门 121.15" "31.89"' '"盐城" "120.13" "33.38"' '"上海" "121.48" "31.22"' '"厦门" "118.1" "24.46"')
#获取数组长度
echo ${#city[@]}
len=$((${#city[@]}-1))
echo $len
#死循环，随机数0-数组长度
while :
do
    rand=`shuf -i0-${len} -n1`
    echo $rand
    echo ${city[${rand}]}
c=(${city[${rand}]})
ci=${c[0]}
cj=${c[1]}
cw=${c[2]}
  req=$(cat /proc/sys/kernel/random/uuid)
date=`date -d 'day' +%Y%m%d`
ts=`date -d $date +%s`
userid=`shuf -i1-100000 -n1`
action=`shuf -i0-1 -n1`
    echo "logger:>>>>$req,$ts,$userid,${ci},${cj},${cw},${action}"
    sleep 3
done

2. 日志采集

日志采集使用Flume，将日志文件数据采集到Kafka中，

[code class="properties"]# 监控本地文件写到kafka
a1.sources = r1
a1.channels = c1
a1.sinks = k1

#设置source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /root/test_flume_file/test_flume_file

#设置Sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = test
a1.sinks.k1.kafka.bootstrap.servers = 192.168.33.4:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.k1.kafka.producer.compression.type = snappy

#设置channels
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

#连接
a1.sources.r1.channels = c1
#注意，是channel
a1.sinks.k1.channel = c1
[/code]

3. KafkaStream进行数据清洗

这一步主要是对日志数据进行清洗，过滤掉不符合规范的日志，

过滤后的日志


requestid,userid,ts,城市，经度，维度，操作(0浏览 1购买)

public static void main(String[] args) {

        String from = "log";
        String to = "process";
        Properties properties = new Properties();

        // 设置application id
        properties.setProperty(StreamsConfig.APPLICATION_ID_CONFIG,"stream-tmall");
        // 设置kafka地址
        properties.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.33.3:9092,192.168.33.4:9092,192.168.33.5:9092");

        // 创建拓扑结构
        Topology topology = new Topology();
        // 构建数据来源，数据处理逻辑，数据去向
        topology.addSource("SOURCE",from)
                .addProcessor("PROCESS", ()-> new LogProcesser(), "SOURCE")
                .addSink("SINK", to, "PROCESS");
        //过时
//        TopologyBuilder builder = new TopologyBuilder();
//        builder.addSource("SOURCE",from)
//                .addProcessor("PROCESS", ()->new LogProcesser(), "SOURCE")
//                .addSink("SINK",to);
//
//
//        // 创建kafkastream
        KafkaStreams streams = new KafkaStreams(topology,properties);
        // 开启流处理
        streams.start();
        System.out.println("kafkaStream is start!!!");


    }
    
    

public class LogProcesser implements Processor<byte[],byte[]> {

    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {

        this.context = context;
    }

    @Override
    public void process(byte[] key, byte[] value) {

        // 核心流程,处理日志
        String line = new String(value);
        if (line.contains("logger:>>>>")){
            System.out.println("LogProcess process data:" + line);
            String[] split = line.split("logger:>>>>");
            // 转发
            context.forward("LogProcess".getBytes(), split[1].trim().getBytes());
        }
        context.commit();
    }

    @Override
    public void punctuate(long timestamp) {

    }

    @Override
    public void close() {

    }
}

  将Kafka log topic中的数据消费，将日志数据进行格式化，写到Process topic中

4. SparkStreaming数据实时处理

我们实时部分，是消费Kafka中process topic中的数据，每一条数据都是今天用户的操作，所以我们将每一条日志的城市取出来，放到realtime topic中，等待后序消费，把数据推送到前端进行实时展示(这部分没做，页面不好写)


public class SparkStreamingProcesser {

    private static final String brokers = "192.168.33.3:9092,192.168.33.4:9092,192.168.33.5:9092";
    private static final String group_id = "tmall_online";
    private static final List<String> topic = new ArrayList<String>(Arrays.asList("process"));

    private static final String toTopic = "realtime";


    public static void main(String[] args) {
        //1. 得到spark上下文
        SparkConf conf = new SparkConf().setAppName("tmall_online").setMaster("local[*]");

        //2. 创建sparkstreamingcontext。每隔2钟会处理一次收集到的数据
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));

        //3. 创建kafka的参数
        HashMap<String, Object> kafkaParams = new HashMap<String, Object>();

        // 设置kafka集群地址
        kafkaParams.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        // 设置消费者组
        kafkaParams.put(ConsumerConfig.GROUP_ID_CONFIG, group_id);
        // 设置key反序列化类
        kafkaParams.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        kafkaParams.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);


        // 4. 通过参数创建一个kafka stream
        JavaInputDStream<ConsumerRecord<Object, Object>> stream = KafkaUtils.createDirectStream(
                jssc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(topic, kafkaParams));

        // 5.获取数据并处理
        JavaDStream<String> map = stream.map(msg-> msg.value().toString());


        JavaDStream<Object> map1 = map.map(x -> {
            KafkaProducerUtils producerUtils = new KafkaProducerUtils();
            String[] split = x.split(",");
            System.out.println("接收到：" + split[1]+","+split[2]);
            producerUtils.sendMessage(toTopic, split[3]+","+split[4]+","+split[5]);
            return "a";
        });
        map1.print();


        jssc.start();

        try {
            jssc.awaitTermination();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            jssc.close();
        }

    }


}

5. 日志持久化

我们需要将用户操作日志持久化到HDFS中，我们将格式化之后的数据采集到HDFS，所以我们使用Flume将Kafka中process topic的数据采集到HDFS

[code class="properties"]# 监控kafka写到hdfs
a1.sources = r1
a1.channels = c1
a1.sinks = k1

#设置source

a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.batchSize = 500
a1.sources.r1.batchDurationMillis = 2000
a1.sources.r1.kafka.bootstrap.servers = 192.168.33.3:9092,192.168.33.4:9092,192.168.33.5:9092
a1.sources.r1.kafka.topics = process
a1.sources.r1.kafka.consumer.group.id = c_flume

#设置Sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /user/customfile/tmall/dt=%Y%m%d
a1.sinks.k1.hdfs.filePrefix = hadoop1_%H_events_
a1.sinks.k1.hdfs.fileSuffix=.log
a1.sinks.k1.hdfs.rollSize = 102400000
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.rollInterval = 3600
a1.sinks.k1.hdfs.idleTimeout = 5400
#是否使用本地时间戳
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#设置文件相关 文件类型为纯文本
a1.sinks.k1.hdfs.fileType = DataStream
#设置channels
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 1000

#连接
a1.sources.r1.channels = c1
#注意，是channel
a1.sinks.k1.channel = c1
[/code]

6. 报表项目

当每天数据采集到HDFS之后，我们需要T+1处理这些日志，产生报表，我们这里主要产生两个指标
  1. 各省活跃用户
  2. 各省销售量
这里我们使用Scala编写Spark程序，做报表，并没有写到mysql，将csv数据写到hdfs中，后序可以导入mysql，电脑带不起了

[code class="scala"]object ReportStatistics {

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

    val spark = SparkSession.builder()
      .appName("report_statistics")
      .enableHiveSupport()
      .getOrCreate()


    val date = args(0)
    //1. 读取前一天数据

    import spark.implicits._

    val df = spark.sparkContext.textFile(s"/user/custom/tmall/dt=$date")
      .filter(x=>{
        val strings: Array[String] = x.split(",")
        strings.length == 7
      })
      .map(x=>{
        val strings: Array[String] = x.split(",")
        (strings(0),strings(1),strings(2),strings(3),strings(6))
      }).toDF("requestid", "ts", "userid", "provice", "action")


    df.cache().createOrReplaceTempView("t1")
    // 统计各省用户日活
    spark.sql(
      s"""
         |select count(distinct(userid)) as users,provice from t1 group by provice
         |union all
         |select count(distinct(userid)) as users,"all" as provice
       """.stripMargin).repartition(1).write
      .mode(SaveMode.Overwrite).csv(s"/user/data/reporting/tmall/dt=$date/users")

    // 统计各省销售量

    spark.sql(
      s"""
         |select count(requestid) as cnt,provice from t1 where action=0 group by provice
         |union all
         |select count(requestid) as cnt from t1 where action=0
       """.stripMargin).repartition(1).write
      .mode(SaveMode.Overwrite).csv(s"/user/data/reporting/tmall/dt=$date/sales")
  }

}
[/code]

总结

项目现在就是做到现在这样，后面还有Azkaban任务调度，Flume将数据采集到ElasticSearch中，用于试试查询，还有web显示页面。

显示的话，报表基本做完，一个web项目读取mysql展示就行，如果有想完善的，可以完善一下，

项目中遇到的问题
  1. Flume将数据采集到HDFS的时候产生了大量的小文件，后面调整了配置参数
源码看这里:  https://gitee.com/zhangqiye/tmall

如有问题，可以加群:  552113611


----------------------------
原文链接:https://www.jianshu.com/p/8fc127944e4a



程序猿的技术大观园：www.javathinker.net 		网站系统异常

返回首页 → 网站系统异常

系统异常信息
Request URL: http://www.javathinker.net/WEB-INF/lybbs/jsp/topic.jsp?postID=1157

java.lang.NullPointerException

如果你不知道错误发生的原因，请把上面完整的信息提交给本站管理人员。