• 数据抽取平台pydatax介绍--实现和项目使用


       数据抽取平台pydatax实现过程中,有2个关键点:

        1、是否能在python3中调用执行datax任务,自己测试了一下可以,代码如下:
        这个str1就是配置的shell文件     

    try:
       result = os.popen(str1).read()
    except Exception as e:
      print(e)

      2、是否能获取datax执行后的信息:用来捕获执行的情况和错误信息

             上面执行后的result就包含了datax的执行信息,对信息进行筛选,就可以获得

       pydatax的表设计 

            在上面的2个关键点解决后,其他问题就比较简单,设计相关的表:

    1
    2
    3
    4
    5
    6
    datax_config   datax抽取表的模板配置(源表名,目标表名,模板id,抽取的字段,抽取条件(增量,全量,特殊),抽取时间,执行顺序等)
    datax_config_repair   datax的出错修复表,结构和datax_config一样,用于datax出错后,修复数据用
    datax_etl_error    datax的etl的报错信息(非异常字符的报错)
    datax_json   datax的模板id配置(全量和增量2个模板文件名)
    datax_log   datax运行抽取表的执行信息(是否执行完成,抽取行数,速度,读出行数,流量等)
    datax_row_error  datax执行中,字段有异常字符的报错信息

     pydatax在项目中使用

           项目1: 直接配置datax的模板json,从oracle 11g抽取到postgresql中,

                         因postgresql中会对"0x"这些异常字符报错,如oracle中字段有这样字段,必须在抽取字段使用:

                        使用 replace(name,chr(0),'\'\'') as name 来代替 以前的字段 name

           项目2: 客户有9个分公司,用的ERP有9套,有9个库,不同版本,抽取的同一个表字段长度有不一样,字段可能有多有少,客户ERP核心分公司ERP几个月后有大版本升级。

                         因项目2中:数据仓库使用的GreePlum,datax的驱动用的是gpdbwriter-v1.0.4-hashdata.jar,该驱动自动删除"0x"非法字符,就不存在该错误

                         不可能写9个抽取json模板,再抽取,只能原有json模板上修改

                         字段长度不同: 取9个库的最大值,作为目标表字段的字段长度

                         字段个数不同:   取其一个核心分公司库表为基础建表,其他8个库表,如果有就保留,没有就字段数据为NULL,每次执行查询取出8个库的字段:                         

    复制代码
    # 获取分公司库该表的字段,如对比核心库表字段的缺失,使用null as 字段替换,如果多余则废弃,
    # 字段对比以核心库为标准
    def get_org_src_columns(src_columns,org_name,tab_name):
        src_columns = src_columns
        # 分公司字段
        org_cols = get_org_cols(org_name,tab_name)
        lst = src_columns.split(",")
        cols1 = (org_cols + ',')
        src_columns1 = (src_columns + ',')
        for i in lst:
          str1 =i.strip() + ','  # 去掉空格,对比使用,字段名+',',这样避免有重复前缀的字段名,导致误判
          if (cols1.find(str(','+str1)) == -1):
            src_columns1 = src_columns1.replace(str(','+str1), ',NULL as ' + str1)
        return src_columns1.rstrip(',')
    
    # 获取分公司库的表的字段用','合并成一个字符串
    def get_org_cols(org_name,tab_name):
        conn = ora_conn()
        cur = conn.cursor()
        cols=""
        sql="select WM_CONCAT(COLUMN_NAME) cols from (SELECT  COLUMN_NAME FROM  all_tab_columns WHERE " \
            " OWNER=upper('"+org_name+"') and  table_name =upper('"+tab_name+"') order by COLUMN_ID asc) t ";
        cur.execute(sql)
        datas = cur.fetchall()
        for row in datas:
          cols= str(row[0])
        return cols;
    复制代码

           修改json模板支持同时抽取9个数据库,修改的9个库同时抽取oracle数据到greeplum全量json模板,见下载文件的:oracle_gp_table_df_job.json:  

    复制代码
        src_table_columns=row.get("src_table_column")
        # 其他8家分公司库
        src_table_columns_fz=get_org_src_columns(src_table_columns,"FZ",src_table_name)
        src_table_columns_jcg=get_org_src_columns(src_table_columns,"JCG",src_table_name)
        src_table_columns_ks=get_org_src_columns(src_table_columns,"KS",src_table_name)
        src_table_columns_qzdf=get_org_src_columns(src_table_columns,"QZDF",src_table_name)
        src_table_columns_sdsht=get_org_src_columns(src_table_columns,"SDSHT",src_table_name)
        src_table_columns_wfjx=get_org_src_columns(src_table_columns,"WFJX",src_table_name)
        src_table_columns_wst=get_org_src_columns(src_table_columns,"WST",src_table_name)
        src_table_columns_std=get_org_src_columns(src_table_columns,"STD",src_table_name)
    
    
        str1 = "/usr/bin/python /opt/module/datax/bin/datax.py /opt/module/datax/job/json/"+etl_mode+" -p  \" " \
               " -Dsrc_table_name='"+src_table_name+"' " \
               " -Ddes_table_name='"+des_table_name+"' " \
               " -Dsplit_pk_field='"+split_pk_field+"'   " \
               " -Drelation='"+relation+"' " \
               " -Dcondition='"+dcondition+"' " \
               " -Dsrc_table_columns='"+src_table_columns+"' " \
               " -Dsrc_table_columns_fz='" + src_table_columns_fz + "' " \
               " -Dsrc_table_columns_jcg='" + src_table_columns_jcg + "' " \
               " -Dsrc_table_columns_ks='" + src_table_columns_ks + "' " \
               " -Dsrc_table_columns_qzdf='" + src_table_columns_qzdf + "' " \
               " -Dsrc_table_columns_sdsht='" + src_table_columns_sdsht + "' " \
               " -Dsrc_table_columns_wfjx='" + src_table_columns_wfjx + "' " \
               " -Dsrc_table_columns_wst='" + src_table_columns_wst + "' " \
               " -Dsrc_table_columns_std='" + src_table_columns_std + "' " \
               " -Ddes_table_columns='"+des_table_columns+"' \" "
    复制代码

          这样修改后,就可以同时抽取9个库的数据,同时配置时,只需要配置核心库的相关字段等数据即可!  

       执行的datax_log表数据:

     

        说明: 1,该平台没有可视化页面的后台管理系统,如果加上后台管理系统,就更完美,但目前是足够使用的!

          DATAX的GreenPlum驱动plugin下载:  

                      https://files.cnblogs.com/files/zping/gpdbwriter.rar?t=1708999240&download=true

         pydatax源码下载地址:

                     https://files.cnblogs.com/files/zping/pydatax.rar?t=1708755764&download=true

        相关文档:

    1.     数据抽取平台pydatax介绍--实现和项目使用
    2.     数据抽取平台pydatax介绍
  • 相关阅读:
    今日睡眠质量记录82分
    【业务架构】什么是价值实现——转型、项目和领导力
    【SLAM】 前端-视觉里程计之相对位姿估计
    【GraphQL】使用Hot Chocolate和.NET 6构建GraphQL应用
    使用Spring Boot和MyBatis访问数据库
    Go 语言服务自动收集线上问题现场
    JuiceFS分布式文件系统源码分析(Java层)
    python LeetCode 刷题记录 35
    多肽标签Avi Tag,GLNDIFEAQKIEWHE
    最新SparkAI创作系统V2.6.2/ChatGPT网站系统H5源码+微信公众号版+AI绘画系统源码/支持GPT联网提问/支持Prompt应用
  • 原文地址:https://www.cnblogs.com/zping/p/18008554