数据中台数据准备

码客说

发布于 2023-03-06 15:53:00

3890

发布于 2023-03-06 15:53:00

文章被收录于专栏：码客

Mysql

创建库

create DATABASE zdb_mysql;

创建表

CREATE TABLE `zdb_mysql`.`t_user`  (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'ID',
  `name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL COMMENT '名字',
  `age` int(11) NULL DEFAULT NULL COMMENT '年龄',
  `height` float NULL DEFAULT NULL COMMENT '身高',
  `birthday` datetime NULL DEFAULT NULL COMMENT '生日',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 3 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;

清空表

TRUNCATE TABLE t_user;

测试数据

INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (1, '蒋震南', 66, 199.49, '2017-09-09 08:38:52');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (2, '', 58, 178.59, '2015-07-15 16:29:07');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (3, null, 45, 169.9, '2003-03-07 18:14:17');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (4, '蒋睿', 97, 164.7, '2022-12-25 11:23:49');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (5, '戴晓明', 6, 189.35, '2008-06-13 03:47:30');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (6, '贾岚', 3, 189.77, '2007-02-07 10:00:37');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (7, '秦安琪', 16, 186.77, '2013-02-13 17:38:43');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (8, '戴璐', 79, 194.92, '2011-09-30 12:02:10');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (9, '宋岚', 86, 156.4, '2004-11-24 22:49:32');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (10, '方安琪', 1, 190.34, '2005-11-30 14:55:37');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (11, '段秀英', 47, 174.5, '2005-04-16 05:46:14');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (12, '雷致远', 36, 169.78, '2002-02-14 07:12:07');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (13, '吕宇宁', 4, 152.15, '2014-10-22 06:10:36');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (14, '韩璐', 71, 176.85, '2000-02-16 16:19:47');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (15, '方子韬', 25, 177.47, '2016-11-16 05:57:48');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (16, '毛詩涵', 71, 199.42, '2021-04-07 19:29:51');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (17, '邓岚', 25, 196.41, '2017-04-05 22:59:59');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (18, '熊安琪', 100, 167.65, '2021-08-18 08:11:03');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (19, '戴岚', 97, 157.19, '2011-06-10 07:20:06');
INSERT INTO `t_user` (`id`, `name`, `age`, `height`, `birthday`) VALUES (20, '莫震南', 54, 174.35, '1992-07-29 10:00:01');

Oracle

不区分大小写。

数据类型

字符型：varchar、char、nchar、nvarchar、long（在数据库中是以ASCII码的格式存储的）

数字型：number、float（表示整数和小数）

日期类型：date、timestamp（存放日期和时间）

其他数据类型：blob、clob、bfile

服务启动与停止

停止服务

sqlplus /nolog
conn /as sysdba
#停止 Oracle 服务
shutdown immediate;
quit

#停止监听器 
lsnrctl stop

启动服务

#启动监听
lsnrctl start
sqlplus /nolog
connect /as sysdba
#启动数据实例
startup
#启动过程中发现内存不够 执行
mount -o remount,size=3G /dev/shm/

创建表空间

临时表空间

创建临时表空间
create temporary tablespace ZDB_TEMP
tempfile '/data/tools/oracle/app/oradata/zdb_temp.dbf'
size 256m
autoextend on
next 500m maxsize 20000m
extent management local;

表空间

-- 创建表空间
create tablespace ZDB datafile '/data/tools/oracle/app/oradata/zdb.dbf'
size 512M reuse autoextend on next 40M maxsize unlimited default storage(
  initial 128k next 128k minextents 2 maxextents unlimited
);
-- 重命名表空间
alter tablespace ZDB rename to MDB;

注意

一个用户可以有多个表空间，一个表空间可以有多个用户，但是为了方便，一个用户只设置一个表空间，这样在执行SQL的时候就不用再指定表空间了。

建用户

create user zhangjian identified by zhangjian
default tablespace ZDB
temporary tablespace temp;

指定表空间给用户

alter user zhangjian default tablespace ZDB;

修改密码

alter user zhangjian identified by zhangjian;

删除用户

--查看用户的连接状态
select username,sid,serial# from v$session; 
--找到要删除用户的sid和serial并杀死
alter system kill session '74,91'; 
--删除用户，及级联关系也删除掉
drop user zhangjian cascade;

Oracle赋权限

grant connect,resource,dba to zhangjian;

删除表

DROP TABLE T_USER;

创建表

CREATE TABLE T_USER (
  "ID" NUMBER NOT NULL ,
  "NAME" VARCHAR2(100) NULL ,
  "AGE" NUMBER NULL,
  "HEIGHT" FLOAT NULL,
  "BIRTHDAY" DATE NULL,
  PRIMARY KEY ("ID")
);

在表空间下建表

CREATE TABLE T_USER (
  "ID" NUMBER NOT NULL ,
  "NAME" VARCHAR2(100) NULL ,
  "AGE" NUMBER NULL,
  "HEIGHT" FLOAT NULL,
  "BIRTHDAY" DATE NULL,
  PRIMARY KEY ("ID")
) tablespace ZDB;

删除数据

delete from T_USER;

插入数据

INSERT INTO "T_USER" ("ID", "NAME", "AGE", "HEIGHT", "BIRTHDAY") VALUES ('1', '陶杰宏', '56', '152.8', TO_DATE('2014-08-07 00:29:45', 'SYYYY-MM-DD HH24:MI:SS'));
INSERT INTO "T_USER" ("ID", "NAME", "AGE", "HEIGHT", "BIRTHDAY") VALUES ('2', '陆杰宏', '98', '195.11', TO_DATE('2022-02-24 17:41:52', 'SYYYY-MM-DD HH24:MI:SS'));
INSERT INTO "T_USER" ("ID", "NAME", "AGE", "HEIGHT", "BIRTHDAY") VALUES ('3', '史子韬', '15', '198.32', TO_DATE('2018-10-01 15:55:30', 'SYYYY-MM-DD HH24:MI:SS'));
INSERT INTO "T_USER" ("ID", "NAME", "AGE", "HEIGHT", "BIRTHDAY") VALUES ('4', '任晓明', '37', '174.93', TO_DATE('2011-09-17 19:02:11', 'SYYYY-MM-DD HH24:MI:SS'));
INSERT INTO "T_USER" ("ID", "NAME", "AGE", "HEIGHT", "BIRTHDAY") VALUES ('5', '谭震南', '100', '186.07', TO_DATE('2020-01-07 23:05:04', 'SYYYY-MM-DD HH24:MI:SS'));
INSERT INTO "T_USER" ("ID", "NAME", "AGE", "HEIGHT", "BIRTHDAY") VALUES ('6', '苏安琪', '79', '179.35', TO_DATE('2008-01-19 16:32:21', 'SYYYY-MM-DD HH24:MI:SS'));
INSERT INTO "T_USER" ("ID", "NAME", "AGE", "HEIGHT", "BIRTHDAY") VALUES ('7', '吕璐', '37', '173.74', TO_DATE('2010-03-08 14:12:57', 'SYYYY-MM-DD HH24:MI:SS'));
INSERT INTO "T_USER" ("ID", "NAME", "AGE", "HEIGHT", "BIRTHDAY") VALUES ('8', '崔子异', '36', '178.07', TO_DATE('2007-02-02 02:22:19', 'SYYYY-MM-DD HH24:MI:SS'));
INSERT INTO "T_USER" ("ID", "NAME", "AGE", "HEIGHT", "BIRTHDAY") VALUES ('9', '许子韬', '76', '168.3', TO_DATE('2014-06-24 02:58:51', 'SYYYY-MM-DD HH24:MI:SS'));
INSERT INTO "T_USER" ("ID", "NAME", "AGE", "HEIGHT", "BIRTHDAY") VALUES ('10', '田詩涵', '51', '184.05', TO_DATE('2021-07-15 15:48:00', 'SYYYY-MM-DD HH24:MI:SS'));

查询数据

select * from T_USER;

Hive

创建库

CREATE DATABASE IF NOT EXISTS zdb;
show databases;
use zdb;

查看表结构

describe t_user_clean;

示例

删除表

drop table t_user01;

创建表

create table t_user01(id bigint,name string,age bigint,height double,birthday string)row format delimited fields terminated by '\001';

以相同结构创建表

create table t_user_trans like t_user01;

创建表

create table t_user01(id bigint,name string,age bigint,height double,birthday string,yxdp_id string,yxdp_process_time string)row format delimited fields terminated by '\001';

清洗的表

create table t_user_clean(id bigint,name string,age bigint,height double,birthday string,yxdp_id string,yxdp_process_time string)row format delimited fields terminated by '\001';

插入数据

插入单条

INSERT into t_user01 values(1, '蒋震南', 66, 199.49, '2017-09-09 08:38:52');

插入多条

INSERT into t_user01 values
(1, '蒋震南', 66, 199.49, '2017-09-09 08:38:52'),
(2, '吴岚', 58, 178.59, '2015-07-15 16:29:07'),
(3, '韩云熙', 45, 169.9, '2003-03-07 18:14:17');

注意

直接使用Hive插入或者在Flink中导入Hive的时候

insert into表名后的字段是不生效的，并且字段的顺序和数量一定要和数据库中保持一致！

下面的SQL

insert into t_user_clean(id,name,age,height,birthday,m1,d1,yxdp_id,yxdp_process_time) select id,name,age,height,birthday,m1,d1,yxdp_id,yxdp_process
_time from UnnamedTable$0

就相当于

insert into t_user_clean select id,name,age,height,birthday,m1,d1,yxdp_id,yxdp_process
_time from UnnamedTable$0

所以说我们就要保证查询时的字段顺序和Hive数据库中的字段顺序一致。

字段处理

获取Hive数据库中的字段

/**
     * 获取所有的字段
     *
     * @param jdbcUrl
     * @param name
     * @param pwd
     * @param tableName
     * @return
     * @throws ClassNotFoundException
     * @throws SQLException
     */
public static LinkedHashMap<String, String> getFields(String jdbcUrl, String name, String pwd, String tableName) throws ClassNotFoundException, SQLException {
  Class.forName("org.apache.hive.jdbc.HiveDriver");
  Connection con = DriverManager.getConnection(jdbcUrl, name, pwd);
  Statement st = con.createStatement();
  String sql = String.format("describe %s", tableName);
  ResultSet resultSet = st.executeQuery(sql);

  LinkedHashMap<String, String> map = new LinkedHashMap<String, String>();
  while (resultSet.next()) {
    String col_name = resultSet.getString("col_name");
    String data_type = resultSet.getString("data_type");
    map.put(col_name, data_type);
  }
  return map;
}

字段要完全按照Hive的字段，所以可能处理的数据缺少我们就要补充缺少的字段

//字段缺少则补充字段
String[] fieldNames = tb01.getSchema().getFieldNames();
for (String fieldName : fieldNameSet) {
  if (!Arrays.asList(fieldNames).contains(fieldName)) {
    String typeStr = fields.get(fieldName);
    switch (typeStr) {
      case "int":
      case "bigint":
        tb01 = tb01.addColumns(call("DefaultValueLongUdf").as(fieldName));
        break;
      case "float":
      case "double":
        tb01 = tb01.addColumns(call("DefaultValueDoubleUdf").as(fieldName));
        break;
      default:
        tb01 = tb01.addColumns(call("DefaultValueStringUdf").as(fieldName));
        break;
    }
  }
}

其中默认值的函数

//添加列时默认值的函数
tableEnv.createTemporarySystemFunction("DefaultValueLongUdf", DefaultValueLongUdf.class);
tableEnv.createTemporarySystemFunction("DefaultValueDoubleUdf", DefaultValueDoubleUdf.class);
tableEnv.createTemporarySystemFunction("DefaultValueStringUdf", DefaultValueStringUdf.class);

DefaultValueLongUdf

import org.apache.flink.table.functions.ScalarFunction;

/**
 * 填充类型的默认值
 */
public class DefaultValueLongUdf extends ScalarFunction {
    public Long eval() {
        return 0L;
    }
}

插入数据

String hiveAllFieldStr = String.join(",", fieldNameSet);
String sqlInsert = String.format("insert into %s select %s from %s where %s=0", tableName, hiveAllFieldStr, tb01.toString(), markDel);

删除数据

Hive表删除数据不能使用DELETE FROM table_name SQL语句

删除所有数据

推荐这种方式比较快(Hive SQL支持，但是Flink SQL中不支持)

truncate table t_user01;

下面的这种方式虽然能删除所有数据，但是不推荐，运行比较慢(Flink SQL中的批模式支持，流模式不支持)。

insert overwrite table t_user01 select * from t_user01 where 1=0;

注意流模式不支持

Streaming mode not support overwrite。

删除部分数据

当需要删除某一条数据的时候，我们需要使用 insert overwrite

释义：就是用满足条件的数据去覆盖原表的数据，这样只要在where条件里面过滤需要删除的数据就可以了

删除id为1的数据：

insert overwrite table t_user01 select * from t_user01 where id <> 1;

Phoenix

库名，表名，字段名如果没加双引号都会变成大写。添加双引号则区分大小写。

删除库

DROP SCHEMA zdb;

注意：确保该 schema 下的表都已删除，否则该 schema 会删除失败。

创建库

create schema zdb;

使用库

use zdb;

删除表

drop table zdb.tuser;

创建表

CREATE TABLE IF NOT EXISTS zdb.tuser(
  id INTEGER primary key,
  name VARCHAR,
  age INTEGER,
  height FLOAT,
  birthday DATE,
  yxdp_id VARCHAR,
  yxdp_process_time VARCHAR
);

插入数据

upsert into tuser values(1, '蒋震南', 66, 199.49, '2017-09-09 08:38:52');
upsert into tuser(id,name,age,height,birthday) values(1, '蒋震南', 66, 199.49, '2017-09-09 08:38:52');