(时间_工作日_节假日_维度表)

1 需求背景

在大数据分析模块中，我们需要从不同的维度分析主题表，包括常用的公用维度：时间维，地区维度，教育信息维......以及各种各样的业务维度：员工维度，部门维度......，业务维度就是我们从哪些角度去分析业务过程，本文就是做一张常用时间维度表。

时间维表由于是可预见的，因此可以一次性导入未来几十年的，当然对于一些节假日的设置可能只能获取未来一年的，因此可以每年全量更新一次。

2 维表设计

给出时间维度表的建表语句

CREATE DATABASE dim_db DEFAULT CHARACTER SET utf8 DEFAULT COLLATE utf8_general_ci;

drop table dim_db.dim_date;
create table if not exists dim_db.dim_date
(
    udate      varchar(20) comment '日期',
    uyear      varchar(20) comment '年',
    uquarter   varchar(20) comment '季度',
    useason    varchar(20) comment '季节',
    umonth     varchar(20) comment '月',
    uday       varchar(20) comment '日',
    uweek      varchar(20) comment '第几周',
    uweekday   varchar(20) comment '周几：1-周一、2-周二、3-周三、4-周四、5-周五、6-周六、7-周日',
    is_workday varchar(20) comment '是否是工作日:1,0',
    udatetype  varchar(20) comment '节假日类型:工作日，法定上班[还班]，周末，节假日',
    updatedate varchar(20) comment '数据更新日期'
);

这里额外解释三个字段

uquarter: 季度，按照阳历的日期分的，[1,2,3]第一季度,[4,5,6]为第二季度,[7,8,9]为第三季度,[10,11,12]为第四季度。

useason:季节，季节的划分有三种方式，本采用的是节气划分法。

• 天文划分法，以春分、夏至、秋分、冬至为四季的开始；

• 气象划分法，以3月至5月为春季，6月至8月为夏季，9月至11月为秋季，12月至2月为冬季；

• 节气划分法，以立春、立夏、立秋、立冬为四季之始。

udatetype:法定上班指的是因为放假调休周末也要上班的日期，节假日会直接显示日期名称，实现的逻辑大家可以自行更改。

3 实现方式

3.1 安装库

本文实现的方式是基于python的chinese_calendar库和pymysql库将数据导入到mysql，接着生成csv文件导入导hive数据库里面。

• MySQL：5.6.36 本地部署
• chinese_calendar：1.8.0 这是一个基于阿里云开发的中国日历库，每年会更新
• pymysql：1.0.2 连接mysql用的
• python版本：3.7

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple chinesecalendar

pip install pymysql

3.2 实现代码

代码写的不太规范，大家见谅

import chinese_calendar
import pymysql.cursors
import datetime
import pytz


def main():
    # (1)设置生成表中数据的开始和截至日期
    start_date = datetime.date(2010, 1, 1)
    end_date = datetime.date(2023, 12, 31)

    # (2)获取该时间段内所有的日期
    dates = chinese_calendar.get_dates(start_date, end_date)

    # (3)遍历日期,构造sql字符串
    strsql = '''INSERT INTO `dim_date`(`udate`,`uyear`,`uquarter`,`useason`,`umonth`,`uday`,`uweek`,`uweekday`,`is_workday`,`udateType`,`updatedate`)VALUES'''
    is_first_line = True

    # (4)设置生成节气的起始时间和截至日期
    # 因为需要 立春，立夏，立秋，立冬四个字段的名称，所以 [数据项的开始日期和截至日期区间]  应在 [节气的起始时间和截至日期的区间]内
    solar_term_start_date = datetime.date(2009, 11, 7)
    solar_term_end_date = datetime.date(2030, 12, 31)

    season_start_date_list = get_season_start_date_list(solar_term_start_date, solar_term_end_date)

    # get_current_season_index
    index = get_current_season_index(season_start_date_list, start_date)

    # (5)遍历日期集合构造sql
    for date in dates:
        # [1]udate: 日期
        udate = date.__str__()

        # [2]uyear: 年份
        uyear = str(date.year)

        # [3]umonth: 月份
        umonth = str(date.month)

        # [4]uquarter: 季度
        uquarter = get_quarter(date.month)

        # [5]useason: 季节
        useason = ''
        if season_start_date_list[index][0] <= date < season_start_date_list[index + 1][0]:
            useason = season_start_date_list[index][1]
        elif date >= season_start_date_list[index + 1][0]:
            index += 1
            useason = season_start_date_list[index][1]

        # [6]uday: 日
        uday = str(date.day)

        # [7]uweek: 第几周
        # 构造一个指定日期时间，时区[必选]的datetime 对象
        timezone = pytz.timezone('Asia/Shanghai')
        dt = datetime.datetime(date.year, date.month, date.day, tzinfo=timezone)
        uweek = str(int(dt.strftime("%U")) + 1)

        # [7]uweekday: 周几
        uweekday = str(date.isoweekday())
        # is_holiday, holidays = chinese_calendar.get_holiday_detail(date)
        is_workday = '1' if chinese_calendar.is_workday(date) else '0'

        # [9]udatetype: 日期类型
        udatetype = ''
        if is_workday == '1':
            if date.isoweekday() in [6, 7]:
                # udatetype = '法定上班'
                is_holiday, holidays = chinese_calendar.get_holiday_detail(date)
                udatetype = '法定上班-' + get_chinese_name(holidays)
            else:
                udatetype = '工作日'
        else:
            is_holiday, holidays = chinese_calendar.get_holiday_detail(date)
            if holidays is None:
                udatetype = '周末'
            else:
                udatetype = get_chinese_name(str(holidays))

        # [10]updatedate: 更新时间，默认是当天
        updatedate = datetime.date.today().__str__()
        # updatedate = datetime.date(2023, 5, 12).__str__() 手动指定更新日期

        # 判断是否首行
        if is_first_line:
            is_first_line = False
        else:
            strsql += ','

        # 构建单行数据库文本记录
        linerecord = '(\'' + udate + '\',\'' \
                     + uyear + '\',\'' \
                     + uquarter + '\',\'' \
                     + useason + '\',\'' \
                     + umonth + '\',\'' \
                     + uday + '\',\'' \
                     + uweek + '\',\'' \
                     + uweekday + '\',\'' \
                     + is_workday + '\',\'' \
                     + udatetype + '\',\'' \
                     + updatedate + '\')'

        print(linerecord)

        # 连接sql
        strsql += linerecord

    strsql += ';'

    # (4)将生成的数据插入到数据库
    # 连接配置信息
    config = {
        'host': '127.0.0.1',
        'port': 3306,
        'user': 'root',
        'password': 'xxxxxxxxx',
        'db': 'dim_dib',
        'charset': 'utf8',
        'cursorclass': pymysql.cursors.DictCursor,

    }

    # 创建连接
    connection = pymysql.connect(**config)

    # 使用cursor创建游标对象
    cursor = connection.cursor()

    # 执行sql语句
    cursor.execute(strsql)
    connection.commit()

    # 关闭连接
    cursor.close()
    connection.close()



def get_quarter(month):
    '''
    获取月份对应的季度

    :param month: 月份
    :return: 季度
    '''
    if month in [1, 2, 3]:
        return '1'
    elif month in [4, 5, 6]:
        return '2'
    elif month in [7, 8, 9]:
        return '3'
    elif month in [10, 11, 12]:
        return '4'
    else:
        return None


def get_chinese_name(english_name):
    '''
    将英文节日名称映射成中文名称

    :param english_name: str
    :return: str
    '''
    if english_name == "New Year's Day":
        return "元旦"
    elif english_name == "Spring Festival":
        return "春节"
    elif english_name == "Tomb-sweeping Day":
        return "清明"
    elif english_name == "Dragon Boat Festival":
        return "端午"
    elif english_name == "Labour Day":
        return "劳动节"
    elif english_name == "National Day":
        return "国庆节"
    elif english_name == "Mid-autumn Festival":
        return "中秋"
    else:
        return "无效节日"


def get_season_start_date_list(start_date, end_date):
    '''
    返回[每个季节开始的日期和名称]的集合
    目前可求的范围是：[1900, 2100]

    :param start_date: 起始时间
    :param end_date: 结束时间
    :return: list，元素内容：(date,'春季')，或者(date,'夏季')，(date,'秋季')，(date,'冬季')
    '''
    solar_terms = chinese_calendar.get_solar_terms(start_date, end_date)

    date_season_list = []
    for term in solar_terms:
        term_name = term[1]
        udate = term[0]
        if term_name == '立春':
            date_season_list.append((udate, '春季'))
        elif term_name == '立夏':
            date_season_list.append((udate, '夏季'))
        elif term_name == '立秋':
            date_season_list.append((udate, '秋季'))
        elif term_name == '立冬':
            date_season_list.append((udate, '冬季'))

    return date_season_list


def get_current_season_index(season_start_date_list, start_date):
    '''
    获取起始时间对应季节的起始时间

    :param season_start_date_list: 每个季节起始时间的日期集合以及相应的季节名称
    :param start_date: 起始时间
    :return: 起始时间对应季节的起始时间 | 集合的下标
    '''
    for index in range(0,12000):
        if start_date >= season_start_date_list[index][0]:
            return index


if __name__ == '__main__':
    main()

3.3 生成数据和更新

使用程序的时候只需要设置程序最开始的start_date和end_date，更新数据同理，不过后续更新数据的时候应该选择更高版本的chinese_calendar库。

4 生成数据预览

这里贴出来了2010年一整年的数据，datagrip导出来的，已经核对过数据是没有问题的，大家可以和国家发布的节假日安排核对假日信息，手机日历上的节假日信息个别有偏差。