二百二十九、离线数仓——离线数仓Hive从Kafka、MySQL到ClickHouse的完整开发流程

一、目的

为了整理离线数仓开发的全流程，算是温故知新吧

离线数仓的数据源是Kafka和MySQL数据库，Kafka存业务数据，MySQL存维度数据

采集工具是Kettle和Flume，Flume采集Kafka数据，Kettle采集MySQL数据

离线数仓是Hive

目标数据库是ClickHouse

任务调度器是海豚

二、数据采集

（一）Flume采集Kafka数据

1、Flume配置文件

## agent a1
a1.sources = s1
a1.channels = c1
a1.sinks = k1

## configure source s1
a1.sources.s1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.s1.kafka.bootstrap.servers = 192.168.0.27:9092
a1.sources.s1.kafka.topics = topic_b_queue
a1.sources.s1.kafka.consumer.group.id = queue_group
a1.sources.s1.kafka.consumer.auto.offset.reset = latest
a1.sources.s1.batchSize = 1000

## configure channel c1
## a1.channels.c1.type = memory
## a1.channels.c1.capacity = 10000
## a1.channels.c1.transactionCapacity = 1000
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /home/data/flumeData/checkpoint/queue
a1.channels.c1.dataDirs = /home/data/flumeData/flumedata/queue

## configure sink k1
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://hurys23:8020/user/hive/warehouse/hurys_dc_ods.db/ods_queue/day=%Y-%m-%d/
a1.sinks.k1.hdfs.filePrefix = queue
a1.sinks.k1.hdfs.fileSuffix = .log
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = second
a1.sinks.k1.hdfs.rollSize = 1200000000
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.idleTimeout = 60
a1.sinks.k1.hdfs.minBlockReplicas = 1

a1.sinks.k1.hdfs.fileType = SequenceFile
a1.sinks.k1.hdfs.codeC = gzip

## Bind the source and sink to the channel
a1.sources.s1.channels = c1
a1.sinks.k1.channel = c1

2、用海豚调度Flume任务

#!/bin/bash
source /etc/profile

/usr/local/hurys/dc_env/flume/flume190/bin/flume-ng agent -n a1 -f /usr/local/hurys/dc_env/flume/flume190/conf/queue.properties

3、目标路径

（二）Kettle采集MySQL维度数据

1、Kettle任务配置

2、用海豚调度Kettle任务

#!/bin/bash
source /etc/profile

/usr/local/hurys/dc_env/kettle/data-integration/pan.sh -rep=hurys_linux_kettle_repository -user=admin -pass=admin -dir=/mysql_to_hdfs/ -trans=23_MySQL_to_HDFS_tb_radar_lane level=Basic >>/home/log/kettle/23_MySQL_to_HDFS_tb_radar_lane_`date +%Y%m%d`.log 

3、目标路径

三、ODS层

（一）业务数据表

use hurys_dc_ods;

create external table  if not exists  ods_queue(
    queue_json  string
)
comment '静态排队数据表——静态分区'
partitioned by (day string)
stored as SequenceFile
;
--刷新表分区
msck repair table ods_queue;
--查看表分区
show partitions ods_queue;
--查看表数据
select * from ods_queue;

（二）维度数据表

use hurys_dc_basic;

create  external  table  if not exists  tb_device_scene(
    id        int      comment '主键id',
    device_no string   comment '设备编号',
    scene_id  string   comment '场景编号'
)
comment '雷达场景表'
row format delimited fields terminated by ','
stored as  textfile  location '/data/tb_device_scene'
tblproperties("skip.header.line.count"="1") ;
--查看表数据
select * from hurys_dc_basic.tb_device_scene;

四、DWD层

（一）业务数据清洗

1、业务数据的JSON有多层

--1、静态排队数据内部表——动态分区  dwd_queue
create  table  if not exists  dwd_queue(
    device_no    string          comment '设备编号',
    lane_num     int             comment '车道数量',
    create_time  timestamp       comment '创建时间',
    lane_no      int             comment '车道编号',
    lane_type    int             comment '车道类型 0:渠化1:来向2:出口3:去向4:左弯待转区5:直行待行区6:右转专用道99:未定义车道',
    queue_count  int             comment '排队车辆数',
    queue_len    decimal(10,2)   comment '排队长度(m)',
    queue_head   decimal(10,2)   comment '排队第一辆车距离停止线距离（m）',
    queue_tail   decimal(10,2)   comment '排队最后一辆车距离停止线距离（m）'
)
comment '静态排队数据表——动态分区'
partitioned by (day string)
stored as orc
;
--动态插入数据

with t1 as(
select
       get_json_object(queue_json,'$.deviceNo')   device_no,
       get_json_object(queue_json,'$.createTime') create_time,
       get_json_object(queue_json,'$.laneNum')    lane_num,
       get_json_object(queue_json,'$.queueList')  queue_list
from hurys_dc_ods.ods_queue
    )
insert  overwrite  table  hurys_dc_dwd.dwd_queue partition(day)
select
        t1.device_no,
        t1.lane_num,
        substr(create_time,1,19)                                               create_time ,
        get_json_object(list_json,'$.laneNo')                                  lane_no,
        get_json_object(list_json,'$.laneType')                                lane_type,
        get_json_object(list_json,'$.queueCount')                              queue_count,
        cast(get_json_object(list_json,'$.queueLen')   as decimal(10,2))       queue_len,
        cast(get_json_object(list_json,'$.queueHead')  as decimal(10,2))       queue_head,
        cast(get_json_object(list_json,'$.queueTail')  as decimal(10,2))       queue_tail,
        date(t1.create_time) day
from t1
lateral view explode(split(regexp_replace(regexp_replace(queue_list,
                                                '\\[|\\]','') ,   --将json数组两边的中括号去掉
                            '\\}\\,\\{','\\}\\;\\{'),  --将json数组元素之间的逗号换成分号
                  '\\;') --以分号作为分隔符(split函数以分号作为分隔)
          )list_queue as list_json
where  device_no is not null  and create_time is not null and  get_json_object(list_json,'$.queueLen') between 0 and 500
and  get_json_object(list_json,'$.queueHead')  between 0 and 500 and  get_json_object(list_json,'$.queueTail')  between 0 and 500 and  get_json_object(list_json,'$.queueCount') between 0 and 100
group by t1.device_no, t1.lane_num, substr(create_time,1,19), get_json_object(list_json,'$.laneNo'), get_json_object(list_json,'$.laneType'), get_json_object(list_json,'$.queueCount'), cast(get_json_object(list_json,'$.queueLen')   as decimal(10,2)), cast(get_json_object(list_json,'$.queueHead')  as decimal(10,2)), cast(get_json_object(list_json,'$.queueTail')  as decimal(10,2)), date(t1.create_time)
;
--查看分区
show partitions dwd_queue;
--查看数据
select * from dwd_queue
where day='2024-03-11';
--删掉表分区
alter table hurys_dc_dwd.dwd_queue drop partition (day='2024-03-11');

2、业务数据的JSON只有一层

--2、转向比数据内部表——动态分区  dwd_turnratio
create  table  if not exists  dwd_turnratio(
    device_no       string        comment '设备编号',
    cycle           int           comment '转向比数据周期' ,
    create_time     timestamp     comment '创建时间',
    volume_sum      int           comment '指定时间段内通过路口的车辆总数',
    speed_avg       decimal(10,2) comment '指定时间段内通过路口的所有车辆速度的平均值',
    volume_left     int           comment '指定时间段内通过路口的左转车辆总数',
    speed_left      decimal(10,2) comment '指定时间段内通过路口的左转车辆速度的平均值',
    volume_straight int           comment '指定时间段内通过路口的直行车辆总数',
    speed_straight  decimal(10,2) comment '指定时间段内通过路口的直行车辆速度的平均值',
    volume_right    int           comment '指定时间段内通过路口的右转车辆总数',
    speed_right     decimal(10,2) comment '指定时间段内通过路口的右转车辆速度的平均值',
    volume_turn     int           comment '指定时间段内通过路口的掉头车辆总数',
    speed_turn      decimal(10,2) comment '指定时间段内通过路口的掉头车辆速度的平均值'
)
comment '转向比数据表——动态分区'
partitioned by (day string)   --分区字段不能是表中已经存在的数据，可以将分区字段看作表的伪列。
stored as orc                 --表存储数据格式为orc
;
--动态插入数据
--解析json字段、去重、非空、volumeSum>=0
--speed_avg、speed_left、speed_straight、speed_right、speed_turn 等字段保留两位小数
--0<=volume_sum<=1000、0<=speed_avg<=150、0<=volume_left<=1000、0<=speed_left<=100、0<=volume_straight<=1000
--0<=speed_straight<=150、0<=volume_right<=1000、0<=speed_right<=100、0<=volume_turn<=100、0<=speed_turn<=100
with t1 as(
select
        get_json_object(turnratio_json,'$.deviceNo')        device_no,
        get_json_object(turnratio_json,'$.cycle')           cycle,
        get_json_object(turnratio_json,'$.createTime')      create_time,
        get_json_object(turnratio_json,'$.volumeSum')       volume_sum,
        cast(get_json_object(turnratio_json,'$.speedAvg')     as decimal(10,2))    speed_avg,
        get_json_object(turnratio_json,'$.volumeLeft')      volume_left,
        cast(get_json_object(turnratio_json,'$.speedLeft')    as decimal(10,2))    speed_left,
        get_json_object(turnratio_json,'$.volumeStraight')  volume_straight,
        cast(get_json_object(turnratio_json,'$.speedStraight')as decimal(10,2))    speed_straight,
        get_json_object(turnratio_json,'$.volumeRight')     volume_right,
        cast(get_json_object(turnratio_json,'$.speedRight')   as decimal(10,2))    speed_right ,
        case when  get_json_object(turnratio_json,'$.volumeTurn')  is null then 0 else get_json_object(turnratio_json,'$.volumeTurn')  end as   volume_turn ,
        case when  get_json_object(turnratio_json,'$.speedTurn')   is null then 0 else cast(get_json_object(turnratio_json,'$.speedTurn')as decimal(10,2))   end as   speed_turn
from hurys_dc_ods.ods_turnratio)
insert overwrite table hurys_dc_dwd.dwd_turnratio partition (day)
select
       t1.device_no,
       cycle,
       substr(create_time,1,19)              create_time ,
       volume_sum,
       speed_avg,
       volume_left,
       speed_left,
       volume_straight,
       speed_straight ,
       volume_right,
       speed_right ,
       volume_turn,
       speed_turn,
       date(create_time) day
from t1
where device_no is not null and volume_sum between 0 and 1000 and speed_avg between 0 and 150 and volume_left  between 0 and 1000
and speed_left between 0 and 100 and volume_straight between 0 and 1000 and speed_straight between 0 and 150
and volume_right between 0 and 1000 and speed_right between 0 and 100 and volume_turn between 0 and 100 and speed_turn between 0 and 100
group by t1.device_no, cycle, substr(create_time,1,19), volume_sum, speed_avg, volume_left, speed_left, volume_straight, speed_straight, volume_right, speed_right, volume_turn, speed_turn, date(create_time)
;
--查看分区
show partitions dwd_turnratio;
--查看数据
select * from hurys_dc_dwd.dwd_turnratio
where day='2024-03-11';
--删掉表分区
alter table hurys_dc_dwd.dwd_turnratio drop partition (day='2024-03-11');

（二）维度数据清洗

create table if not exists  dwd_radar_lane(
    device_no         string  comment '雷达编号',
    lane_no           string  comment '车道编号',
    lane_id           string  comment '车道id',
    lane_direction    string  comment '行驶方向',
    lane_type         int     comment '车道类型 0渠化，1来向路段，2出口，3去向路段，4路口，5非路口路段，6其他',
    lane_length       float   comment '车道长度',
    lane_type_name    string  comment '车道类型名称'
)
comment '雷达车道信息表'
stored as orc
;
--create table if not exists  dwd_radar_lane  stored as orc as
--加载数据
insert overwrite table  hurys_dc_dwd.dwd_radar_lane
select
device_no, lane_no, lane_id, lane_direction, lane_type,lane_length ,
       case when lane_type='0' then '渠化'
            when lane_type='1' then '来向路段'
            when lane_type='2' then '出口'
            when lane_type='3' then '去向路段'
       end as lane_type_name
from hurys_dc_basic.tb_radar_lane
where lane_length is not null
group by device_no, lane_no, lane_id, lane_direction, lane_type, lane_length
;
--查看表数据
select * from hurys_dc_dwd.dwd_radar_lane;

五、DWS层

create  table  if not exists  dws_statistics_volume_1hour(
    device_no        string         comment '设备编号',
    scene_name       string         comment '场景名称',
    lane_no          int            comment '车道编号',
    lane_direction   string         comment '车道流向',
    section_no       int            comment '断面编号',
    device_direction string         comment '雷达朝向',
    sum_volume_hour  int            comment '每小时总流量',
    start_time       timestamp      comment '开始时间'
)
comment '统计数据流量表——动态分区——1小时周期'
partitioned by (day string)
stored as orc
;
--动态加载数据  --两个一起 1m41s 、 convert.join=false  1m43s、
--注意字段顺序  查询语句中字段顺序与建表字段顺序一致
insert  overwrite  table  hurys_dc_dws.dws_statistics_volume_1hour  partition(day)
select
       dwd_st.device_no,
       dwd_sc.scene_name,
       dwd_st.lane_no,
       dwd_rl.lane_direction,
       dwd_st.section_no,
       dwd_rc.device_direction,
       sum(volume_sum) sum_volume_hour,
       concat(substr(create_time, 1, 14), '00:00') start_time,
       day
from hurys_dc_dwd.dwd_statistics as dwd_st
    right join hurys_dc_dwd.dwd_radar_lane as dwd_rl
              on dwd_rl.device_no=dwd_st.device_no and dwd_rl.lane_no=dwd_st.lane_no
    right join hurys_dc_dwd.dwd_device_scene as dwd_ds
              on dwd_ds.device_no=dwd_st.device_no
    right join hurys_dc_dwd.dwd_scene as dwd_sc
              on dwd_sc.scene_id = dwd_ds.scene_id
    right join hurys_dc_dwd.dwd_radar_config as dwd_rc
              on dwd_rc.device_no=dwd_st.device_no
where dwd_st.create_time is not null
group by dwd_st.device_no, dwd_sc.scene_name, dwd_st.lane_no, dwd_rl.lane_direction, dwd_st.section_no, dwd_rc.device_direction, concat(substr(create_time, 1, 14), '00:00'), day
;
--查看分区
show partitions dws_statistics_volume_1hour;
--查看数据
select * from hurys_dc_dws.dws_statistics_volume_1hour
where day='2024-02-29';

六、ADS层

这里的ADS层，其实就是用Kettle把Hive的DWS层结果数据同步到ClickHouse中，也是一个Kettle任务而已

这样用海豚进行调度每一层的任务，整个离线数仓流程就跑起来了

七、海豚调度任务（除了2个采集任务外）

（一）delete_stale_data（根据删除策略删除ODS层原始数据）

#! /bin/bash
source /etc/profile

nowdate=`date --date='0 days ago' "+%Y%m%d"`
day_30_ago_date=`date -d "30 day ago " +%Y-%m-%d`

#静态排队数据
hadoop fs -test -e /user/hive/warehouse/hurys_dc_ods.db/ods_queue/day=${day_30_ago_date}
if [ $? -ne 0 ]; then
    echo "文件不存在"
else 
    hdfs dfs -rm -r /user/hive/warehouse/hurys_dc_ods.db/ods_queue/day=${day_30_ago_date}
fi

#轨迹数据
hadoop fs -test -e /user/hive/warehouse/hurys_dc_ods.db/ods_track/day=${day_30_ago_date}
if [ $? -ne 0 ]; then
    echo "文件不存在"
else 
    hdfs dfs -rm -r /user/hive/warehouse/hurys_dc_ods.db/ods_track/day=${day_30_ago_date}
fi

#动态排队数据
hadoop fs -test -e /user/hive/warehouse/hurys_dc_ods.db/ods_queue_dynamic/day=${day_30_ago_date}
if [ $? -ne 0 ]; then
    echo "文件不存在"
else 
    hdfs dfs -rm -r /user/hive/warehouse/hurys_dc_ods.db/ods_queue_dynamic/day=${day_30_ago_date}
fi

#区域数据
hadoop fs -test -e /user/hive/warehouse/hurys_dc_ods.db/ods_area/day=${day_30_ago_date}
if [ $? -ne 0 ]; then
    echo "文件不存在"
else 
    hdfs dfs -rm -r /user/hive/warehouse/hurys_dc_ods.db/ods_area/day=${day_30_ago_date}
fi

#事件数据
hadoop fs -test -e /user/hive/warehouse/hurys_dc_ods.db/ods_event/day=${day_30_ago_date}
if [ $? -ne 0 ]; then
    echo "文件不存在"
else 
    hdfs dfs -rm -r /user/hive/warehouse/hurys_dc_ods.db/ods_event/day=${day_30_ago_date}
fi

#删除表分区
hive -e "
use hurys_dc_ods;

alter table hurys_dc_ods.ods_area drop partition (day='$day_30_ago_date');
alter table hurys_dc_ods.ods_event drop partition (day='$day_30_ago_date');
alter table hurys_dc_ods.ods_queue drop partition (day='$day_30_ago_date');
alter table hurys_dc_ods.ods_queue_dynamic drop partition (day='$day_30_ago_date');
alter table hurys_dc_ods.ods_track drop partition (day='$day_30_ago_date')
"

（二）flume（Flume采集Kafka业务数据）

（三）create_database_table（自动创建Hive和ClickHouse的库表）

1、创建Hive库表

#! /bin/bash
source /etc/profile

hive -e "
source  1_dws.sql
"

2、创建ClickHouse库表

#! /bin/bash
source /etc/profile

clickhouse-client --user default --password hurys@123 -d default --multiquery <1_ads.sql

（四）hive_dws（DWS层任务）

#! /bin/bash
source /etc/profile

nowdate=`date --date='0 days ago' "+%Y%m%d"`
yesdate=`date -d yesterday +%Y-%m-%d`

hive -e "
use hurys_dc_dws;

set hive.vectorized.execution.enabled=false;

set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.max.dynamic.partitions.pernode=1000;
set hive.exec.max.dynamic.partitions=2000;
    
            
insert  overwrite  table  hurys_dc_dws.dws_statistics_volume_1hour  partition(day='$yesdate')
select
       dwd_st.device_no,
       dwd_sc.scene_name,
       dwd_st.lane_no,
       dwd_rl.lane_direction,
       dwd_st.section_no,
       dwd_rc.device_direction,
       sum(volume_sum) sum_volume_hour,
       concat(substr(create_time, 1, 14), '00:00') start_time
from hurys_dc_dwd.dwd_statistics as dwd_st
    right join hurys_dc_dwd.dwd_radar_lane as dwd_rl
              on dwd_rl.device_no=dwd_st.device_no and dwd_rl.lane_no=dwd_st.lane_no
    right join hurys_dc_dwd.dwd_device_scene as dwd_ds
              on dwd_ds.device_no=dwd_st.device_no
    right join hurys_dc_dwd.dwd_scene as dwd_sc
              on dwd_sc.scene_id = dwd_ds.scene_id
    right join hurys_dc_dwd.dwd_radar_config as dwd_rc
              on dwd_rc.device_no=dwd_st.device_no
where dwd_st.create_time is not null  and  day= '$yesdate'
group by dwd_st.device_no, dwd_sc.scene_name, dwd_st.lane_no, dwd_rl.lane_direction, dwd_st.section_no, dwd_rc.device_direction, concat(substr(create_time, 1, 14), '00:00')    
"

（五）hive_basic（维度表基础库）

#! /bin/bash
source /etc/profile

hive -e "
set hive.vectorized.execution.enabled=false;

use hurys_dc_basic
"

（六）dolphinscheduler_log（删除海豚日志文件）

#! /bin/bash
source /etc/profile

nowdate=`date --date='0 days ago' "+%Y%m%d"`
yesdate=`date -d yesterday +%Y-%m-%d`

cd  /usr/local/hurys/dc_env/dolphinscheduler/dolphin/logs/

rm -rf dolphinscheduler-api.$yesdate*.log
rm -rf dolphinscheduler-master.$yesdate*.log
rm -rf dolphinscheduler-worker.$yesdate*.log

（七）Kettle_Hive_to_ClickHouse（Kettle采集Hive的DWS层数据同步到ClickHouse的ADS层中）

#!/bin/bash
source /etc/profile

/usr/local/hurys/dc_env/kettle/data-integration/pan.sh -rep=hurys_linux_kettle_repository -user=admin -pass=admin -dir=/hive_to_clickhouse/ -trans=17_Hive_to_ClickHouse_ads_avg_volume_15min level=Basic >>/home/log/kettle/17_Hive_to_ClickHouse_ads_avg_volume_15min_`date +%Y%m%d`.log 

（八）Kettle_MySQL_to_HDFS（Kettle采集MySQL维度表数据到HDFS中）

（九）hive_dwd（DWD层任务）

1、业务数据的JSON有多层

#! /bin/bash
source /etc/profile

nowdate=`date --date='0 days ago' "+%Y%m%d"`
yesdate=`date -d yesterday +%Y-%m-%d`

hive -e "
use hurys_dc_dwd;

set hive.vectorized.execution.enabled=false;

set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.max.dynamic.partitions.pernode=1000;
set hive.exec.max.dynamic.partitions=1500;

with t1 as(
select
       get_json_object(queue_json,'$.deviceNo')   device_no,
       get_json_object(queue_json,'$.createTime') create_time,
       get_json_object(queue_json,'$.laneNum')    lane_num,
       get_json_object(queue_json,'$.queueList')  queue_list
from hurys_dc_ods.ods_queue
where date(get_json_object(queue_json,'$.createTime')) = '$yesdate'
    )
insert  overwrite  table  hurys_dc_dwd.dwd_queue partition(day='$yesdate')
select
        t1.device_no,
        t1.lane_num,
        substr(create_time,1,19)                                               create_time ,
        get_json_object(list_json,'$.laneNo')                                  lane_no,
        get_json_object(list_json,'$.laneType')                                lane_type,
        get_json_object(list_json,'$.queueCount')                              queue_count,
        cast(get_json_object(list_json,'$.queueLen')   as decimal(10,2))       queue_len,
        cast(get_json_object(list_json,'$.queueHead')  as decimal(10,2))       queue_head,
        cast(get_json_object(list_json,'$.queueTail')  as decimal(10,2))       queue_tail
from t1
lateral view explode(split(regexp_replace(regexp_replace(queue_list,
                                                '\\\\[|\\\\]','') ,      --将json数组两边的中括号去掉
                                 '\\\\}\\\\,\\\\{','\\\\}\\\\;\\\\{'),   --将json数组元素之间的逗号换成分号
                   '\\\\;')   --以分号作为分隔符(split函数以分号作为分隔)
          )list_queue as list_json
where  device_no is not null  and  get_json_object(list_json,'$.queueLen') between 0 and 500 and  get_json_object(list_json,'$.queueHead')  between 0 and 500 and  get_json_object(list_json,'$.queueTail')  between 0 and 500 and  get_json_object(list_json,'$.queueCount') between 0 and 100
group by t1.device_no, t1.lane_num, substr(create_time,1,19), get_json_object(list_json,'$.laneNo'), get_json_object(list_json,'$.laneType'), get_json_object(list_json,'$.queueCount'), cast(get_json_object(list_json,'$.queueLen')   as decimal(10,2)), cast(get_json_object(list_json,'$.queueHead')  as decimal(10,2)), cast(get_json_object(list_json,'$.queueTail')  as decimal(10,2))
"

2、业务数据的JSON单层

#! /bin/bash
source /etc/profile

nowdate=`date --date='0 days ago' "+%Y%m%d"`
yesdate=`date -d yesterday +%Y-%m-%d`

hive -e "
use hurys_dc_dwd;

set hive.vectorized.execution.enabled=false;

set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.max.dynamic.partitions.pernode=1000;
set hive.exec.max.dynamic.partitions=1500;

with t1 as(
select
        get_json_object(turnratio_json,'$.deviceNo')        device_no,
        get_json_object(turnratio_json,'$.cycle')           cycle,
        get_json_object(turnratio_json,'$.createTime')      create_time,
        get_json_object(turnratio_json,'$.volumeSum')       volume_sum,
        cast(get_json_object(turnratio_json,'$.speedAvg')     as decimal(10,2))    speed_avg,
        get_json_object(turnratio_json,'$.volumeLeft')      volume_left,
        cast(get_json_object(turnratio_json,'$.speedLeft')    as decimal(10,2))    speed_left,
        get_json_object(turnratio_json,'$.volumeStraight')  volume_straight,
        cast(get_json_object(turnratio_json,'$.speedStraight')as decimal(10,2))    speed_straight,
        get_json_object(turnratio_json,'$.volumeRight')     volume_right,
        cast(get_json_object(turnratio_json,'$.speedRight')   as decimal(10,2))    speed_right ,
        case when  get_json_object(turnratio_json,'$.volumeTurn')  is null then 0 else get_json_object(turnratio_json,'$.volumeTurn')  end as   volume_turn ,
        case when  get_json_object(turnratio_json,'$.speedTurn')   is null then 0 else cast(get_json_object(turnratio_json,'$.speedTurn')as decimal(10,2))   end as   speed_turn
from hurys_dc_ods.ods_turnratio
where date(get_json_object(turnratio_json,'$.createTime')) = '$yesdate'
)
insert overwrite table hurys_dc_dwd.dwd_turnratio partition (day='$yesdate')
select
       t1.device_no,
       cycle,
       substr(create_time,1,19)              create_time ,
       volume_sum,
       speed_avg,
       volume_left,
       speed_left,
       volume_straight,
       speed_straight ,
       volume_right,
       speed_right ,
       volume_turn,
       speed_turn
from t1
where device_no is not null and volume_sum between 0 and 1000 and speed_avg between 0 and 150 and volume_left  between 0 and 1000 and speed_left between 0 and 100 and volume_straight between 0 and 1000 and speed_straight between 0 and 150 and volume_right between 0 and 1000 and speed_right between 0 and 100 and volume_turn between 0 and 100 and speed_turn between 0 and 100
group by t1.device_no, cycle, substr(create_time,1,19), volume_sum, speed_avg, volume_left, speed_left, volume_straight, speed_straight, volume_right, speed_right, volume_turn, speed_turn
"

3、维度数据

#! /bin/bash
source /etc/profile

hive -e "
use hurys_dc_dwd;

set hive.vectorized.execution.enabled=false;

insert overwrite table hurys_dc_dwd.dwd_holiday
select
day, holiday,year
from hurys_dc_basic.tb_holiday
group by day, holiday, year
"

（十）hive_ods（ODS层任务）

#! /bin/bash
source /etc/profile

hive -e "
use hurys_dc_ods;

msck repair table ods_queue;

msck repair table ods_turnratio;

msck repair table ods_queue_dynamic;

msck repair table ods_statistics;

msck repair table ods_area;

msck repair table ods_pass;

msck repair table ods_track;

msck repair table ods_evaluation;

msck repair table ods_event;
"

目前，整个离线数仓的流程大致就是这样，有问题的后面再完善！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/519864.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

JVM专题——垃圾回收

JVM专题——垃圾回收

本文部分内容节选自Java Guide和《深入理解Java虚拟机》, Java Guide地址: https://javaguide.cn/java/jvm/jvm-garbage-collection.html &#x1f680; 基础&#xff08;上&#xff09; → &#x1f680; 基础&#xff08;中&#xff09; → &#x1f680;基础&#xff08;下&…
阅读更多...
2024-04-05 问AI: 介绍一下深度学习中的Leaky ReLU函数

2024-04-05 问AI: 介绍一下深度学习中的Leaky ReLU函数

文心一言 Leaky ReLU&#xff08;Leaky Rectified Linear Unit&#xff09;函数是深度学习领域中的一种激活函数&#xff0c;主要用于解决ReLU&#xff08;Rectified Linear Unit&#xff09;函数在负值区域为零的问题。ReLU函数在输入为正时保持原样&#xff0c;而在输入为负…
阅读更多...
（学习日记）2024.04.07：UCOSIII第三十五节：互斥量实验

（学习日记）2024.04.07：UCOSIII第三十五节：互斥量实验

写在前面&#xff1a; 由于时间的不足与学习的碎片化&#xff0c;写博客变得有些奢侈。 但是对于记录学习&#xff08;忘了以后能快速复习&#xff09;的渴望一天天变得强烈。 既然如此 不如以天为单位&#xff0c;以时间为顺序&#xff0c;仅仅将博客当做一个知识学习的目录&a…
阅读更多...
通过 Cookie、Redis共享Session 和 Spring 拦截器技术，实现对用户登录状态的持有和清理（三）

通过 Cookie、Redis共享Session 和 Spring 拦截器技术，实现对用户登录状态的持有和清理（三）

本篇内容对应 “2.4 生成验证码” 小节 和 “4.7 优化登陆模块”小节 视频链接 1 Kaptcha介绍 Kaotcga是一个生成验证码的工具。 你的网站验证码是什么&#xff1f; 在我们这个牛客论坛项目&#xff0c;验证码分为两部分 给用户看的是图片&#xff0c;用户根据图片上显示的…
阅读更多...
跨境电商独立站是什么？为什么要做独立站？

跨境电商独立站是什么？为什么要做独立站？

跨境电商独立站就是跨境电商自行搭建的销售网站&#xff0c;服务器、域名都是自主购买的&#xff0c;并由跨境电商独立运营与营销推广。 近些年来&#xff0c;各类第三方电商平台虽然流量大&#xff0c;但是随着进驻电商数量的增加&#xff0c;流量竞争也愈发激烈&#xff0c;…
阅读更多...
基于顺序表实现通讯管理系统！(有完整源码！)

基于顺序表实现通讯管理系统！(有完整源码！)

​​​​​​​ 个人主页&#xff1a;秋风起&#xff0c;再归来~ 文章专栏&#xff1a;C语言实战项目 个人格言&#xff1a;悟已往之不谏&#xff0c;知来者犹可追 克心守己&#xff0c;律己则安&#xff01;​​​​​​​ 目录 1、实现思路 ​…
阅读更多...
C语言中strlen函数的实现

C语言中strlen函数的实现

C语言中strlen函数的实现 为了便于和strlen函数区别&#xff0c;以下命令为_strlen。 描述&#xff1a;实现strlen&#xff0c;获取字符串的长度&#xff0c;函数原型如下&#xff1a; size_t strlen(const char *str);_strlen实现&#xff1a; size_t _strlen(const char*…
阅读更多...
彩虹聚合DNS管理系统,附带系统搭建教程

彩虹聚合DNS管理系统,附带系统搭建教程

聚合DNS管理系统&#xff0c;可以实现在一个网站内管理多个平台的域名解析&#xff0c;目前已支持的域名平台有&#xff1a;阿里云、腾讯云、华为云、西部数码、CloudFlare。 本系统支持多用户&#xff0c;每个用户可分配不同的域名解析权限&#xff1b;支持API接口&#xff0…
阅读更多...
武汉星起航：跨境电商领域的领航者，助力全球贸易新篇章

武汉星起航：跨境电商领域的领航者，助力全球贸易新篇章

自2017年以来&#xff0c;武汉星起航一直专注于亚马逊自营店铺&#xff0c;积累了宝贵的经验。2020年正式成立后&#xff0c;公司以跨境电商为核心&#xff0c;致力于为合作伙伴提供深入的合作模式。武汉星起航凭借其卓越的服务和实战经验&#xff0c;已成功助力众多创业者实现…
阅读更多...
基于SpringBoot的“智慧外贸平台”的设计与实现（源码+数据库+文档+PPT)

基于SpringBoot的“智慧外贸平台”的设计与实现（源码+数据库+文档+PPT)

基于SpringBoot的“智慧外贸平台”的设计与实现&#xff08;源码数据库文档PPT) 开发语言&#xff1a;Java 数据库&#xff1a;MySQL 技术&#xff1a;SpringBoot 工具&#xff1a;IDEA/Ecilpse、Navicat、Maven 系统展示 系统总体结构图 平台首页界面图 商品信息界面图 …
阅读更多...
Java8 进阶

Java8 进阶

Java8 进阶 文章目录 Java8 进阶什么是函数式接口&#xff1f;public interface Supplierpublic interface Consumerpublic interface Predicatepublic interface FunctionJava8 特性总结&#xff1a;一、Function<T, R>二、Consumer<T>三、Supplier<T>四、P…
阅读更多...
位运算-191. 位1的个数- 136. 只出现一次的数字

位运算-191. 位1的个数- 136. 只出现一次的数字

位1的个数 已解答 简单 相关标签 相关企业 编写一个函数&#xff0c;输入是一个无符号整数&#xff08;以二进制串的形式&#xff09;&#xff0c;返回其二进制表达式中 设置位 的个数&#xff08;也被称为汉明重量&#xff09;。 示例 1&#xff1a; 输入&#xff1a;n 11 输…
阅读更多...
Linux第4课 Linux的基本操作

Linux第4课 Linux的基本操作

文章目录 Linux第4课 Linux的基本操作一、图形界面介绍二、终端界面介绍 Linux第4课 Linux的基本操作 一、图形界面介绍 本节以Ubuntu系统的GUI为例进行说明&#xff0c;Linux其他版本可自行网搜。 图形系统进入后&#xff0c;左侧黄框内为菜单栏&#xff0c;右侧为桌面&…
阅读更多...
c# 指数搜索（Exponential Search）

c# 指数搜索（Exponential Search）

该搜索算法的名称可能会产生误导&#xff0c;因为它的工作时间为 O(Log n)。该名称来自于它搜索元素的方式。 给定一个已排序的数组和要 搜索的元素 x&#xff0c;找到 x 在数组中的位置。 输入&#xff1a;arr[] {10, 20, 40, 45, 55} x 45 输出&#xff1a;在索…
阅读更多...
检验平台最基本的技术要求有哪几条

检验平台最基本的技术要求有哪几条

检验平台最基本的技术要求通常有以下几条&#xff1a; 系统稳定性&#xff1a;检验平台应具备良好的稳定性&#xff0c;能够长时间运行而不出现系统崩溃或异常情况。 数据安全性&#xff1a;检验平台应具备对数据进行安全存储和传输的能力&#xff0c;确保数据不被非法获取、篡…
阅读更多...
吴恩达机器学习笔记：第 6 周-11机器学习系统的设计(Machine Learning System Design)11.1-11.5

吴恩达机器学习笔记：第 6 周-11机器学习系统的设计(Machine Learning System Design)11.1-11.5

目录 第 6 周 11、 机器学习系统的设计(Machine Learning System Design)11.1 首先要做什么11.2 误差分析11.3 类偏斜的误差度量11.4 查准率和查全率之间的权衡11.5 机器学习的数据 第 6 周 11、 机器学习系统的设计(Machine Learning System Design) 11.1 首先要做什么 在接…
阅读更多...
基于Python的豆瓣电影评分可视化，豆瓣电影评分预测系统

基于Python的豆瓣电影评分可视化，豆瓣电影评分预测系统

博主介绍&#xff1a;✌程序员徐师兄、7年大厂程序员经历。全网粉丝12w、csdn博客专家、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java技术领域和毕业项目实战✌ &#x1f345;文末获取源码联系&#x1f345; &#x1f447;&#x1f3fb; 精彩专栏推荐订阅&#x1f447;…
阅读更多...
2024年阿里云4月服务器有哪些优惠活动？

2024年阿里云4月服务器有哪些优惠活动？

2024年阿里云服务器4月优惠活动有哪些&#xff1f;4月份最新优惠活动有99计划云服务器99元一年、学生服务器、游戏服务器优惠、云服务器精选特惠、高校计划优惠券300元、阿里云服务器免费试用等活动。4月云服务器最新优惠价格2核2G3M带宽99元一年、2核4G5M带宽199元一年&#x…
阅读更多...
AcWing 312. 乌龟棋（每日一题）

AcWing 312. 乌龟棋（每日一题）

原题链接&#xff1a;312. 乌龟棋 - AcWing题库 小明过生日的时候&#xff0c;爸爸送给他一副乌龟棋当作礼物。 乌龟棋的棋盘只有一行&#xff0c;该行有 N 个格子&#xff0c;每个格子上一个分数&#xff08;非负整数&#xff09;。 棋盘第 1 格是唯一的起点&#xff0c;第…
阅读更多...
vue + koa + Sequelize + 阿里云部署 + 宝塔：宝塔数据库连接

vue + koa + Sequelize + 阿里云部署 + 宝塔：宝塔数据库连接

之前文章已经介绍了宝塔上传前后端代码并部署&#xff0c;不清楚的请看这篇文章&#xff1a; vue koa 阿里云部署 宝塔&#xff1a;宝塔前后端部署 下面是宝塔创建数据库&#xff1a; 我用的 koa Sequelize 连接的数据库&#xff0c;Sequelize 非常适合前端使用&#xf…
阅读更多...
最新文章

Copyright @ 2022~2023