美文网首页大数据那些事
数据仓库日期维表怎么做

数据仓库日期维表怎么做

作者: 乌拉乌拉儿 | 来源:发表于2020-04-16 14:32 被阅读0次

    最近终于有点时间了,东拼西凑的设计了一个日期维表,也是数仓设计时比较常用的。表结构如下:

    dt string COMMENT '日历日期',
    lunar_dt string COMMENT '农历日期',
    calendar_year int COMMENT '日历年',
    quarter_of_year int COMMENT '季度',
    month_of_year int COMMENT '月份',
    week_of_year int COMMENT '一年第几周',
    dt_of_week int COMMENT '一周第几天',
    dt_of_month int COMMENT '一个月第几天',
    dt_of_year int COMMENT '一年第几天',
    is_last_day_of_month boolean COMMENT '是否一个月最后一天',
    is_weekend boolean COMMENT '是否周末日',
    is_holiday boolean COMMENT '是否节假日',
    holiday_cnt tinyint COMMENT '节日天数',
    holiday_type tinyint COMMENT '日历天类型:1:工作日;2:周末(和假期连着按假期,因节假日调休上班按工作日算:3:元旦 4:春节 5:清明节 6:劳动节:7:端午节 8:中秋节 9:国庆'

    实际落表代码没有写,只写了数据获取部分,具体实现代码如下,欢迎大佬指点~

    #!/usr/bin/python
    # -*- coding: utf-8 -*-
    
    import sys
    import requests
    import json
    import datetime
    import time
    import calendar
    import lunardate
    import csv
    
    reload(sys)                      # reload 才能调用 setdefaultencoding 方法
    sys.setdefaultencoding('utf-8')
    
    holiday_dict = {}
    
    # 判断二月闰月
    def isLeapYear(year):
        return calendar.monthrange(year, 2)[1] == 29
    
    # 季度信息
    def quarter(month):
        return (month - 1) / 3 + 1
    
    # 周信息
    def get_week_of_year(dt):
        return dt.isocalendar()[1]
    
    def get_day_of_year(dt):
        return dt.timetuple().tm_yday
    
    def get_day_of_week(day):
        return day.weekday() + 1 % 7
    # 返回是否最后一天
    def last_day_of_month(year, month, day):
        return day == calendar.monthrange(year, month)[1]
    
    def get_is_weekend(day):
        return day.weekday() == 5 or day.weekday() == 6
    
    def get_holiday_dict(year, month):
        params = {
            'resource_id': 6018,
            'ie': 'utf8',
            'oe': 'utf8',
            'format': 'json',
            'tn': 'baidu'
        }
        query_arg = "%d年%d月"
        params['query'] = query_arg % (year, month)
        res = requests.get(url='https://sp0.baidu.com/8aQDcjqpAAV3otqbppnN2DJv/api.php?', params=params).text
        # 取括号内的
        res = res[res.find("{"): res.rfind("}") + 1]
    
        js = json.loads(res,encoding='utf-8')
    
        holiday_list = js.get('data')[0].get('holiday')
    
        for i in holiday_list:
            list_list = i.get('list')
            name = i.get('name')
            day_cnt = len(list_list)
            for x in list_list:
                status = x.get('status')
                if status == 2:
                    day_cnt -= 1
    
            for j in list_list:
                dt = j.get('date')
                status = j.get('status')
                item_info = (status,name,day_cnt)
                holiday_dict[dt] = item_info
    
    # 获取阴历日期
    def get_lunar_dt(dt):
        ymd = dt.split('-')
        lunar = lunardate.LunarDate.fromSolarDate(int(ymd[0]), int(ymd[1]), int(ymd[2]))
        return "%d-%02d-%02d" % (lunar.year, lunar.month, lunar.day)
    
    
    def get_is_holiday(day):
        dt = day.strftime("%Y-%-m-%-d")
        item = holiday_dict.get(dt)
        if item is not None :
            return int(item[0]) == 1
        else:
            return False
    
    def is_work_day(day,is_weekend):
        dt = day.strftime("%Y-%-m-%-d")
        item = holiday_dict.get(dt)
        if item is not None :
            # 如果在holiday列表里但是状态等于2
            return int(item[0]) == 2
        else:
            return not (is_weekend)
    
    def get_holiday_cnt(day):
        dt = day.strftime("%Y-%-m-%-d")
        item = holiday_dict.get(dt)
        if item is not None and int(item[0]) == 1:
            return int(item[2])
        else:
            return 0
    
    def get_holiday_type(day,is_weekend):
        dt = day.strftime("%Y-%-m-%-d")
        item = holiday_dict.get(dt)
        if item is not None :
            if int(item[0]) == 1:
                if str(item[1])  == '元旦'  :
                    return 3
                elif str(item[1])  == '春节' :
                    return 4
                elif str(item[1]) == '清明节' :
                    return 5
                elif str(item[1])  == '劳动节' :
                    return 6
                elif str(item[1])  == '端午节' :
                    return 7
                elif str(item[1])  == '中秋节' :
                    return 8
                elif str(item[1])  == '国庆节':
                    return 9
                else:
                    return 99
            else:
                return 1
    
        else:
            if is_weekend:
                return 2
            else:
                return 1
        return 1
    
    def write_csv(data_row):
        path  = "aa.csv"
        with open(path,'a+') as f:
            csv_write = csv.writer(f)
            csv_write.writerow(data_row)
    
    if __name__ == '__main__':
        start_date = datetime.datetime(2020,1,1)
        end_date = datetime.datetime(2020,1,31)
        day = start_date
        get_holiday_dict(2020,1)
    
        while(day <= end_date):
            dt = day.strftime("%Y-%m-%d")
            lunar_dt = get_lunar_dt(dt)
    
            calendar_year = day.year
            quarter_of_year = quarter(day.month)
            month_of_year = day.month
            week_of_year = get_week_of_year(day)
            dt_of_week = get_day_of_week(day)
            dt_of_month = day.day
            dt_of_year = get_day_of_year(day)
            is_last_day_of_month = last_day_of_month(calendar_year,month_of_year,dt_of_month)
            is_weekend = get_is_weekend(day)
            is_holiday = get_is_holiday(day)
            holiday_type = get_holiday_type(day,is_weekend)
            holiday_cnt = get_holiday_cnt(day)
            data_raw = [dt,lunar_dt,calendar_year,quarter_of_year,month_of_year,week_of_year,dt_of_week,dt_of_month,dt_of_year,is_last_day_of_month,is_weekend,is_holiday,holiday_type,holiday_cnt]
            print data_raw
            write_csv(data_raw)
            # print '日期:'+dt+ '年份'+str(calendar_year)+'Q'+str(quarter_of_year)+'月份'+str(month_of_year)+'周'+str(week_of_year)+'星期'+str(dt_of_week)+' 月的第'+str(dt_of_month)+' 年的第'+str(dt_of_year)+str(is_weekend)
    
            day = day + datetime.timedelta(days=1)
    

    相关文章

      网友评论

        本文标题:数据仓库日期维表怎么做

        本文链接:https://www.haomeiwen.com/subject/pusxbctx.html