当前位置：首页 > article >正文

python创建udf函数步骤

article 2024/11/15 13:33:49

一、目标

实现一个函数，传入两个datetime类型的参数，返回double类型的工作日天数

二、思路

如何计算差值？
如果开始时间和结束时间在同一天：实现同 datediff(@end, @start, ‘ss’) / 86400.0
如果开始时间和结束时间在不同天：将时间分为三部分

a.
开始时间 - 开始时间当天的24点
b.
结束时间当天的0点 - 结束时间
c.
开始时间当天的24点 - 结束时间当天的0点的工作日天数

在这里插入图片描述

三、实现

1、下载三方包

2、编译生成WHEEL包

将第三方包解压到本地，在系统的命令行窗口，切换路径至setup.py文件所在文件夹
在系统的命令行窗口，执行如下命令编译生成WHEEL包

huahuo@HUAHUOdeMacBook-Pro ~ % cd Downloads 
huahuo@HUAHUOdeMacBook-Pro Downloads % cd chinesecalendar-1.8.1
huahuo@HUAHUOdeMacBook-Pro chinesecalendar-1.8.1 % python setup.py bdist_wheel

修改后缀为ZIP格式
在这里插入图片描述
重命名为

3、上传资源

hdfs dfs -put /path/to/chinese_holiday.zip /hdfs/path/to/work/chinese_holiday.zip

4、新建Python资源

chinese_holiday.py

● 计算 end 和 start 之间的总天数，如果发生异常（比如传入的不是有效的 datetime 对象），它将返回 None。
● 然后，从 start 日期的第二天开始迭代，一直到 end 日期的前一天。对于每一天，如果 is_holiday(start_date) 返回 True（表示这一天是节假日），则将 total_days 减一。is_holiday 函数来自于 chinese_calendar 模块，用于判断指定的日期是否为中国的节假日。
● 在迭代期间，如果发生异常（比如无法从 chinese_calendar 模块导入 is_holiday 函数或其他原因），它将返回当前计算的 total_days。
● 如果没有异常，最终返回的 total_days 将是 end 和 start 日期之间的工作日数（即剔除了节假日的天数）。
总结来说，这个自定义函数的目的是计算两个日期之间的中国工作日天数，忽略周末和公共假期。


class is_chinese_holiday(object):
    def __init__(self):
        import sys
        sys.path.insert(0, 'work/chinese_holiday.zip')

    def evaluate(self, thedate):
        from chinese_calendar import get_holiday_detail, is_in_lieu
        from datetime import datetime
        from json import dumps
        info = {'is_holiday': 'Invalid'}
        try:
            target = datetime.strptime(thedate, '%Y%m%d').date()
            detail = get_holiday_detail(target)
            info['is_holiday'] = str(detail[0])
            info['is_in_lieu'] = str(is_in_lieu(target))
            if (detail[0]):
                info['holiday_name'] = str(detail[1])
                return dumps(info)
            else:
                return dumps(info)
        except:
            return dumps(info)



class get_chinese_workdays(object):
    def __init__(self):
        import sys
        sys.path.insert(0, 'work/chinese_holiday.zip')

    def evaluate(self, end, start):
        from chinese_calendar import is_holiday
        from datetime import datetime, timedelta
        try:
            total_days = (end - start).total_seconds() / 86400.0
        except:
            return None
        try:
            start_date = start.date() + timedelta(days=1)
            while start_date < end.date():
                total_days -= int(is_holiday(start_date))
                start_date += timedelta(days=1)
            return total_days
        except:
            return total_days

5、新建Maxc函数

将你的 Python UDF 脚本上传到 Hive 服务器上，或者放置在一个 Hive 能够访问到的位置，确保 Hive 服务器上已经安装了 Python，并且你的脚本具有执行权限。
你需要在 Hive 会话中注册这个 Python 脚本作为一个 UDF。这可以通过 ADD FILE 命令来完成，然后使用 TRANSFORM 关键字调用这个脚本：
– 将 Python 脚本添加到 Hive
ADD FILE /path/to/chinese_holiday.py;

– 创建一个临时的自定义函数
CREATE TEMPORARY FUNCTION get_chinese_workdays AS ‘chinese_holiday.get_chinese_workdays’;

– 使用 UDF 转换数据
SELECT get_chinese_workdays (column_name)
FROM your_table;

查看全文

http://www.kler.cn/a/232873.html