热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

推荐引擎数据导入模块的实现

毕设做到后半部分,需要实现将用户在一段时间(比如1天)内产生的新数据导入HDFS的功能,这样数据仓库中的数据才能和数据库中的数据同步在新建了一个PyDev项目后,需要如下操作(拣最

毕设做到后半部分,需要实现将用户在一段时间(比如1天)内产生的新数据导入HDFS的功能,这样数据仓库中的数据才能和数据库中的数据同步

在新建了一个PyDev项目后,需要如下操作(拣最主要的写):

 

模块的环境变量:

# -*- coding:UTF-8 -*-
#
!/usr/bin/python # FileName:pro_env.py

#*************************************************** # 项目的路径 PROJECT_DIR = "/usr/local/EclipseProjects/MyBI" # 项目配置文件的路径 PROJECT_CONF_DIR = PROJECT_DIR + "/conf/" # 项目第三方库的路径 PROJECT_LIB_DIR = PROJECT_DIR + "/lib" # 项目临时文件的路径 PROJECT_TMP_DIR = PROJECT_DIR + "/temp" #*************************************************** # Hadoop的安装路径 HADOOP_HOME = "/usr/local/hadoop/" # Hadoop的命令路径 HADOOP_PATH = HADOOP_HOME + "bin/" # HIVE的安装路径 HIVE_HOME = "/opt/hive-0.9.0/" # HIVE的命令路径 HIVE_PATH = HIVE_HOME + "bin/" # Sqoop的安装路径 SQOOP_HOME = "/opt/Sqoop/" # Sqoop的命令路径 SQOOP_PATH = SQOOP_HOME + "bin/" #*************************************************** # Java的安装路径 Java_HOME = "/usr/lib/jvm/jdk1.7.0_75"

配置文件:

导入模块的配置文件主要的目的是告诉Sqoop,导入哪些表,怎么导入,我暂时需要一张表,新建一个XML文件Import.xml,type="add"表示增量导入

xml version="1.0" encoding="UTF-8"?>
<root>
    <task type="add">
        <table>ModifyRecordstable>
    task>
root>

需要对每张表进行更细一步的配置,新建ModifyRecords.xml

xml version="1.0" encoding="UTF-8"?>
<root>
    <sqoop-shell type="import">
        <param key="connect">jdbc:mysql://localhost:3306/Recommendparam>
        <param key="username">${username}param>
        <param key="password">${password}param>
        <param key="target-dir">/user/hadoop/Recommend/$dtparam>
        <param key="query">‘select userID,movieID,pref from Recommend.ModifyRecords where modifyDate$flag"\$CONDITIONS" and $CONDITIONS‘param>
        <param key="m">1param>
        <param key="fields-terminated-by">‘,‘param>
    sqoop-shell>
root>

剩下的工作就是解析配置文件:

# -*- coding:UTF-8 -*-
#!/usr/bin/python
# FileName:import.py
from com.utls.pro_env import PROJECT_CONF_DIR
import time
from com.utls.sqoop import SqoopUtil
import xml.etree.ElementTree as ET

# 其中dt为昨天的日期,将由调度模块传入
def resolve_conf(dt):
    
    # 获得配置文件名
    conf_file = PROJECT_CONF_DIR + "Import.xml"
    
    # 解析配置文件
    xml_tree = ET.parse(conf_file)
    # 获得task元素
    tasks = xml_tree.findall(./task)
    
    for task in tasks:
        # 获得导入类型,增量导入或者全量导入
        import_type = task.attrib["type"]
        
        # 获得表名集合
        tables = task.findall(./table)
        
        # 用来保存待执行的Sqoop命令的集合
        cmds = []

        # 迭代表名集合,解析表配置文件
        for i in range(len(tables)):
            # 表名
            table_name = tables[i].text
            # 表配置文件名
            table_conf_file = PROJECT_CONF_DIR + table_name + ".xml"
            
            # 解析表配置文件
            xmlTree = ET.parse(table_conf_file)
            
            # 获取sqoop-shell节点
            sqoopNodes = xmlTree.findall("./sqoop-shell")
            
            # 获取sqoop-shell节点
            sqoop_cmd_type = sqoopNodes[0].attrib["type"]
            # 获取
            praNodes = sqoopNodes[0].findall("./param")
            
            # 用来保存param信息的字典
            cmap = {}
            
            for i in range(len(praNodes)):
                # 获得key属性的值
                key = praNodes[i].attrib["key"]
                # 获得param标签中间的值
                value = praNodes[i].text
                # 保存到字典中
                cmap[key] = value
                
            # 首先组装成sqoop命令头
            command = "sqoop " + sqoop_cmd_type
                
            # 如果为全量导入
            if(import_type == "all"):
                # query的查询条件为
                import_cOndition= dt
                flag = "<"
            # 如果为增量导入
            elif (import_type == "add"):
                # query的查询条件为=dt
                import_cOndition= dt
                flag = "="
            else:
                raise Exception
                
            # #迭代字典将param的信息拼装成字符串
            for key in cmap.keys():
                    
                value = cmap[key]
                    
                # 如果不是键值对形式的命令选项
                if(value == None or value == "" or value == " "):
                    value = ""
                    
                # 将query的CONDITIONS替换为查询条件
                if(key == "query"):
                    value = value.replace("\$CONDITIONS", import_condition)
                    value = value.replace("$flag", flag)
                        
                # 将导入分区替换为传入的时间
                if(key == "target-dir"):
                    value = value.replace("$dt", dt)
                    
                # 拼装为命令
                if key == "fields-terminated-by":
                    command += " --" + key + " " + value
                else:
                    command += " --" + key + " " + value + "\\" + "\n"
                
            # 将命令加入至待执行的命令集合
            cmds.append(command)
        
    return cmds

# Python模块的入口:main函数
if __name__ == __main__:
    
    # 调度模块将昨天的时间传入
    dt = time.strftime("%Y-%m-%d", time.localtime(time.time()))
    # 解析配置文件,获得sqoop命令集合
    cmds = resolve_conf(dt)
    
    # 迭代集合,执行命令
    for i in range(len(cmds)):
        cmd = cmds[i]
        
        # 执行导入过程
        SqoopUtil.execute_shell(cmd)

拼装出来的命令如下:

sqoop import --username xxxx --target-dir /user/hadoop/Recommend/2015-04-26 --m 1 --connect jdbc:mysql://localhost:3306/Recommend --query select userID,movieID,pref from Recommend.ModifyRecords where modifyDate="2015-04-26" and $CONDITIONS --password xxxx --fields-terminated-by ,

最后新建一个模块(不过当然写在import.py的main函数之前...),编写一个类,为该类编写一个函数,目的是用Python调用Sqoop命令:

#!/usr/bin/python
# FileName sqoop.py
# -*- coding:UTF-8 -*-
import os
class SqoopUtil(object):
    ‘‘‘
    sqoop operation
    ‘‘‘
    def __init__(self):
        pass
    
    @staticmethod
    def execute_shell(shell):
        print shell
        os.system(shell)
        

推荐引擎数据导入模块的实现


推荐阅读
  • 本文介绍了SIP(Session Initiation Protocol,会话发起协议)的基本概念、功能、消息格式及其实现机制。SIP是一种在IP网络上用于建立、管理和终止多媒体通信会话的应用层协议。 ... [详细]
  • 二维码的实现与应用
    本文介绍了二维码的基本概念、分类及其优缺点,并详细描述了如何使用Java编程语言结合第三方库(如ZXing和qrcode.jar)来实现二维码的生成与解析。 ... [详细]
  • 我的读书清单(持续更新)201705311.《一千零一夜》2006(四五年级)2.《中华上下五千年》2008(初一)3.《鲁滨孙漂流记》2008(初二)4.《钢铁是怎样炼成的》20 ... [详细]
  • empty,isset首先都会检查变量是否存在,然后对变量值进行检测。而is_null只是直接检查变量值,是否为null,因此如果变量未定义就会出现错误!检测一个变量是否是null ... [详细]
  • 在1995年,Simon Plouffe 发现了一种特殊的求和方法来表示某些常数。两年后,Bailey 和 Borwein 在他们的论文中发表了这一发现,这种方法被命名为 Bailey-Borwein-Plouffe (BBP) 公式。该问题要求计算圆周率 π 的第 n 个十六进制数字。 ... [详细]
  • 在日常生活中,支付宝已成为不可或缺的支付工具之一。本文将详细介绍如何通过支付宝实现免费提现,帮助用户更好地管理个人财务,避免不必要的手续费支出。 ... [详细]
  • publicclassBindActionextendsActionSupport{privateStringproString;privateStringcitString; ... [详细]
  • 本文介绍了如何通过C#语言调用动态链接库(DLL)中的函数来实现IC卡的基本操作,包括初始化设备、设置密码模式、获取设备状态等,并详细展示了将TextBox中的数据写入IC卡的具体实现方法。 ... [详细]
  • OBS Studio自动化实践:利用脚本批量生成录制场景
    本文探讨了如何利用OBS Studio进行高效录屏,并通过脚本实现场景的自动生成。适合对自动化办公感兴趣的读者。 ... [详细]
  • 本文详细介绍了C++中的构造函数,包括其定义、特点以及如何通过构造函数进行对象的初始化。此外,还探讨了转换构造函数的概念及其在不同情境下的应用,以及如何避免不必要的隐式类型转换。 ... [详细]
  • 理解浏览器历史记录(2)hashchange、pushState
    阅读目录1.hashchange2.pushState本文也是一篇基础文章。继上文之后,本打算去研究pushState,偶然在一些信息中发现了锚点变 ... [详细]
  • Windows操作系统提供了Encrypting File System (EFS)作为内置的数据加密工具,特别适用于对NTFS分区上的文件和文件夹进行加密处理。本文将详细介绍如何使用EFS加密文件夹,以及加密过程中的注意事项。 ... [详细]
  • 解决Visual Studio Code中PHP Intelephense误报问题
    PHP作为一种高度灵活的编程语言,其代码结构可能导致Intelephense插件在某些情况下报告不必要的错误或警告。自1.3.3版本起,Intelephense引入了多个配置选项,允许用户根据具体的工作环境和编程风格调整这些诊断信息的显示。 ... [详细]
  • 心理学经典:《思考致富》
    《思考致富》是由美国著名成功学大师拿破仑·希尔撰写的一部重要著作,该书基于希尔长达20年的深入研究和访谈,探讨了个人成功的核心要素。书中不仅揭示了成功的关键,还提供了一系列实用的方法和策略。 ... [详细]
  • 本文将详细介绍如何在二进制和十六进制之间进行准确的转换,并提供实际的代码示例来帮助理解这一过程。 ... [详细]
author-avatar
ik82jht
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有