[大数据]数据埋点到hdfs

项目资源如下：

https://gitee.com/martina-zhong/mal

项目如何构建请参考：

从零开始搭建springboot项目骨架

项目首页

这是一个mall商城项目，本节主要是完成用户对商品加入购物车的信息数据埋点写入到hdfs。

思路：当用户点击购物车，后台获取到用户以及商品的json数据，上传到hdfs。

一、步骤

1.添加写入hdfs的事件

小tips：
1、ctrl+f可搜索
2、按住ctrl点击事件可跳转到响应事件函数的代码块位置

经过观察，商品详情页在info.html，找到“加入购物车”按钮，发现点击按钮出发了一个事件。

找到这个事件，加入点击写入hdfs的函数

代码：

//埋点    把我们需要的数据     埋到hdfs
$.get("addHDFS?productId="+productId,function (data) {
    //
})

2.找作用域

目的是确定是哪一个用户点击的加入购物车，获取该用户的相关信息（如电话，地址等）

在登录页面用户信息提交到表单


通过HttpSession session可以获取到用户信息。

用如下代码可以获取到用户实体类信息

User user = (User) session.getAttribute("user");

3.写hdfsDao层

写hdfs的api和加载配置

package priv.jesse.mall.dao;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;

import java.io.IOException;
import java.net.URI;
import java.util.Properties;

public class HDFSDao {
    //hdfs的API
    static FileSystem fileSystem;
    static Properties properties;
    static {
        //properties
         properties = new Properties();
        //加载配置文件
        try {
            properties.load(HDFSDao.class.getClassLoader().getResourceAsStream("application.properties"));
        } catch (IOException e) {
            e.printStackTrace();
        }
        Configuration conf = new Configuration();
        conf.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
        conf.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
        conf.set("dfs.client.use.datanode.hostname","true");
        try {
             fileSystem = FileSystem.get(URI.create(properties.getProperty("spring.hdfs.url")), conf, "root");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 对外提供一个获取Properties的方法
     */
    public static Properties getProperties(){
        return properties;
    }

    /**
     * 对外提供一个获取fileSystem的方法
     */
    public static FileSystem getFileSystem(){
        return fileSystem;
    }
}

注意： 如果是配置的云服务器，则需要加这行

conf.set(“dfs.client.use.datanode.hostname”,“true”);

如果配置的是本机hadoop则不用

4.写service层

先写hdfsService接口：

package priv.jesse.mall.service;

public interface HDFSService {
    /**
     * 把埋点的json数据写入hdfs
     */
     void logToHDFS(String line);
}

写hdfsServiceImpl实现该接口（主要是把获取的json数据写入hdfs）：

package priv.jesse.mall.service.impl;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.springframework.stereotype.Service;
import priv.jesse.mall.dao.HDFSDao;
import priv.jesse.mall.service.HDFSService;
import priv.jesse.mall.utils.TimeUtils;

import java.io.IOException;
import java.util.Properties;

@Service
public class HDFSServiceImpl implements HDFSService {

    /**
     * 把json数据埋到hdfs
     * @param line
     */
    @Override
    public  void logToHDFS(String line) {
        //实现具体逻辑
        FileSystem fileSystem = HDFSDao.getFileSystem();
        Properties properties = HDFSDao.getProperties();
        //操作   数据==》HDFS
        try {
//            //获取hdfs的输出流     内容再次写入被覆盖
            String path=properties.getProperty("spring.hdfs.shoppingLog")+TimeUtils.getHDFSTimes()+".log";
            //把输出流提出去
            FSDataOutputStream outputStream=null;
            //if判断
            if(fileSystem.exists(new Path(path))){
                //存在
                 outputStream = fileSystem.append(new Path(path));
            }else{
               // 不存在
                 outputStream = fileSystem.create(new Path(path));
            }
            //appen   InputStream in, OutputStream out, Configuration conf, boolean close
            Configuration configuration = new Configuration();
            configuration.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
            configuration.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
            configuration.set("dfs.client.use.datanode.hostname","true");
            //把一个String类型变成inputStream
            IOUtils.copyBytes(org.apache.commons.io.IOUtils.toInputStream(line+"\r\n"),outputStream,configuration,true);
            //关流
            outputStream.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        new HDFSServiceImpl().logToHDFS("svfjsgfeedghdkghdhgd");
    }
}

这里调用了utils的工具包

package priv.jesse.mall.utils;

import java.text.SimpleDateFormat;
import java.util.Date;

public class TimeUtils {
    //添加当前时间

    public  static String getTimes(){
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        String format1 = format.format(new Date());
        return format1;
    }

    public  static String getHDFSTimes(){
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");

        String format1 = format.format(new Date());
        return format1;
    }

}

5.写controller层（实际上就是写加入hdfs的控制）

HttpSession session用来获取作用域（即登录时用户的实体类信息）

@ResponseBody
@RequestMapping("/addHDFS")
public void addToHDFS(int productId,HttpSession session) throws Exception {
    //调用service层方法，查询到商品的信息
    Product product = productService.findById(productId);
    //开始设计我们需要埋的数据     用户的userID从哪里来
    User user = (User) session.getAttribute("user");
    //String name, String phoneNumber, int productId, String title,
    // Double shopPrice, String times
    ShoppingLogs shoppingLogs = new ShoppingLogs(user.getName(),user.getPhone(),
            product.getId(),product.getTitle(),product.getShopPrice(),TimeUtils.getTimes());
    //输出 ShoppingLogs{name='pengfei zhang', phoneNumber='15035451125', productId=10, title='Apple iPhone X (A1865) 64GB 银色 移动联通电信4G手机',
    // shopPrice=8888.0, times='2022-05-11 08:55:21'}
    //为了将来处理方便  我们需要把埋好的数据变成json
    String line = JSON.toJSONString(shoppingLogs);
    System.out.println(line);

    //怎么把这个json落地到hdfs
    hdfsService.logToHDFS(line);
}

注意： 转为json要导入依赖

<dependency>
	<groupId>com.alibaba</groupId>
	<artifactId>druid-spring-boot-starter</artifactId>
	<version>1.1.10</version>
</dependency>

二、遇到的问题

1.虚拟机连不上网

ifconfig发现设备是eth1

解决方案：
1、关闭防火墙

chkconfig iptables stop

2、修改配置文件

详情见：
hadoop保姆级安装教程

看具体信息（UUID在里面）

ip addr

打开配置文件，编辑

vi /etc/sysconfig/network-scripts/ifcfg-eth0

我的打开是这样的

修改为（各个参数详情见上述配置hadoop链接）：

DEVICE=“eth0”
BOOTPROTO=“static”
HOSTNAME=“quickstart.cloudera”
HWADDR=“00:0C:29:D3:4D:5C”
IPV6INIT=“no”
MTU=“1500”
NM_CONTROLLED=“yes”
ONBOOT=“yes”
TYPE=“Ethernet”
IPADDR=x.x.x.x
NETMASK=255.255.255.0
GETEWAY=192.168.179.2
DNS1=192.168.179.2
DNS2=114.114.114.114
UUID=“f127faf3-7a1f-4bca-a404-0119665a8459”
~

因为设备是eth1，所以要有eth1的文件，移动并重命名

mv /etc/sysconfig/network-scripts/ifcfg-eth0 /etc/sysconfig/network-scripts/ifcfg-eth1

重启网络

service network restart

不行再重启虚拟机

2、本机连不上虚拟机的网

这里我使用的连接器是MobaXterm_Personal_10.5，本机ping不通，自然就连接不上

在VMware里编辑网络配置（在虚拟机关机状态下）
编辑->更改设置->勾选

启动虚拟机

打开连接工具，输入虚拟机ip和用户名，ok之后输入密码

3.
hadoop解决Exception in thread “main” java.lang.NoClassDefFoundError: org/apache/hadoop/yarn/util/Apps
（其实就是没有导入yarn相应的jar包和lib包，导入就好了）
https://blog.csdn.net/zhangge360/article/details/51181794

报错原因：windows没有装hadoop，有些资源访问不到

解决办法：
①下载一个hadoop的包，解压（解压到英文路径下）
进入hadoop目录下

把bin里面的这两个文件复制

复制到C:\Windows\System32
②修改环境变量

找到编辑系统环境变量->环境变量

新建一个系统环境变量


注意： 名字不能有错，变量值是安装hadoop的路径

编辑path变量

新建一个变量，名字为%HADOOP_HOME%\bin

点击确定，确定，确定

5、

原因：主机没添加云服务器的ip和主机名

解决方案：
到C:\Windows\System32\drivers\etc，修改hosts

==注意：==这里的hosts文件一般是只读的，无法修改，所以右击，选择属性

取消勾选只读，确定，就可以修改了

在最后一行添加云服务器的ip，空格，云服务器的主机名，保存

这个问题就解决了

欢迎各位小伙伴探讨私聊哦，需要hadoop等等资源的可以私我，谢谢支持~

开发者涨薪指南

48位大咖的思考法则、工作方式、逻辑体系