实时即未来，车联网项目之将数据落地到文件系统和数据库【三】

#!/bin/bash

dt=`date -d '1 days ago' +'%Y%m%d'`
tableName=$1

ssh node03 `/export/server/hive/bin/hive -e "use maynor_ods;alter table ${tableName} add partition(dt=${dt}) location 'hdfs://node01:8020/apps/warehouse/ods.db/${tableName}/${dt}"`

# 使用shell 脚本
alter table maynor_src add partition (dt="20210922") location "/apps/hive/warehouse/ods.db/maynor_src/20210922";

package cn.maynor.streaming.sink;

import cn.maynor.streaming.entity.maynorDataObj;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;

/**
 * Author maynor
 * Date 2021/9/22 10:02
 * Desc 将每条车辆的数据直接写入到 Hive 中
 */
public class SaveErrorDataHiveSink extends RichSinkFunction<maynorDataObj> {
    //定义 logger
    private static final Logger logger = LoggerFactory.getLogger(SaveErrorDataHiveSink.class);
    //2.创建有参构造方法，参数包括数据库名和表名
    //定义变量
    private String dbName;
    private String tableName;
    //定义连接对象和statement对象
    private Connection conn = null;
    private Statement statement = null;
    //构造方法
    public SaveErrorDataHiveSink(String _dbName,String _tableName){
        this.dbName = _dbName;
        this.tableName = _tableName;
    }

    //3.重写open方法进行Hive连接的初始化
    @Override
    public void open(Configuration parameters) throws Exception {
        //3.1 将JDBC驱动 org.apache.hive.jdbc.HiveDriver 加载进来
        //获取全局参数
        ParameterTool parameterTool = (ParameterTool) getRuntimeContext()
                .getExecutionConfig()
                .getGlobalJobParameters();
        //获取当前上下文中 hive 的驱动
        Class.forName(parameterTool.getRequired("hive.driver"));
        //3.2 设置JDBC连接Hive的连接器，端口为10000
        conn = DriverManager.getConnection(
                parameterTool.getRequired("hive.url"),
                parameterTool.getRequired("hive.user"),
                parameterTool.get("hive.password")
        );
        //3.3 创建Statement
        statement = conn.createStatement();
        //3.4 定义 schemaAndTableExists 实现库不存在创建库，表不存在创建表
        Boolean flag = schemaAndTableExists(dbName,tableName,statement);
        if(flag){
            logger.info("当前数据库和表初始化成功！");
        }else{
            logger.warn("请检查数据库和表！");
        }
    }

    //5.重写cloese方法 关闭连接
    @Override
    public void close() throws Exception {
        if(!statement.isClosed())statement.close();
        if(!conn.isClosed())conn.close();
    }

    //4.重写invoke将每条数据
    @Override
    public void invoke(maynorDataObj value, Context context) throws Exception {
        //4.1 编写SQL将数据插入到表中
        // insert into maynor_error values('11111');
        StringBuffer buffer = new StringBuffer();
        buffer.append("INSERT INTO "+tableName);
        buffer.append(" VALUES('");
        buffer.append(value.getErrorData()+"'");
        //4.2 执行statement.executeUpdate 将数据直接落地到Hive表
        statement.executeUpdate(buffer.toString());
    }

    //6.定义 schemaAndTableExists 方法 create database if not exists库或表, execute，选择数据库

    /**
     * 初始化数据库和数据表，如果初始化成功返回 true，否则 false
     * @param dbName
     * @param tableName
     * @param statement
     * @return
     */
    private Boolean schemaAndTableExists(String dbName, String tableName, Statement statement) {
        //数据库是否存在
        Boolean flag = true;
        try{
            //初始化数据库
            String createDBSQL="create database if not exists "+dbName;
            boolean executeDB = statement.execute(createDBSQL);
            if(executeDB){
                logger.info("当前数据库创建成功");
                flag = true;
            }else{
                logger.info("当前数据库已经存在");
                flag = true;
            }
            //初始化数据表
            String createTableSQL = "use "+tableName+";create table if not exists "+tableName+" (json string) partition by dt" +
                    " row formatted delimited field terminate by '\t' location '/apps/hive/warehouse/ods.db/maynor_error'";
            boolean executeTable = statement.execute(createTableSQL);
            if(executeTable){
                logger.info("当前数据库表创建成功");
                flag = true;
            }else{
                logger.info("当前数据表已经存在");
                flag = true;
            }
        }catch (Exception ex){
            logger.warn("初始化失败！");
            flag = false;
        }
        return flag;
    }
}

# 首先开启 hdfs ，zookeeper 
/export/server/hbase/start-hbase.sh

hbase shell

# 查看hbase所有表
list
# 查看namespace（数据库）
list_namespace
# 创建数据表
hbase(main):005:0> create 'maynor_src','cf'
# 查看表中的数据
scan 'maynor_src'

//1.创建 SrcDataToHBaseSink类继承 RichSinkFunction
//2.创建一个有参数-表名的构造方法
//3.重写open方法
//3.1 从上下文获取到全局的参数
//3.2 设置hbase的配置，Zookeeper Quorum集群和端口和TableInputFormat的输入表
//3.3 通过连接工厂创建连接
//3.4 通过连接获取表对象
//4.重写close方法
//4.1 关闭hbase 表和连接资源
//5. 重写 invoke 方法，将读取的数据写入到 hbase
//5.1 setDataSourcePut输入参数value，返回put对象
//6. 实现 setDataSourcePut 方法
//6.1 如何设计rowkey VIN+时间戳翻转
//6.2 定义列簇的名称
//6.3 通过 rowkey 实例化 put
//6.4 将所有的字段添加到put的字段中

BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(tableName));
        params.writeBufferSize(10 * 1024 * 1024L);
        params.setWriteBufferPeriodicFlushTimeoutMs(5 * 1000L);

//5.1 setDataSourcePut输入参数value，返回put对象
        try {
            Put put = setDataSourcePut(value);
            mutator.mutate(put);
            //5.2 指定时间内的数据强制刷写到hbase
            mutator.flush();
        }catch (Exception ex){
            logger.error("写入到hbase失败："+ex.getMessage());
        }

alter 'maynor_src', { NAME => 'cf', DATA_BLOCKs_ENCODING => 'FAST_DIFF' }

create 'maynor_src',{NAME => 'cf',COMPRESSION => 'gz'}create 'maynor_src_snappy', { NAME => 'cf', COMPRESSION => 'SNAPPY' }