抽取oracle数据到mysql数据库的实现过程

load data infile "alarm_hist_inc.csv" into table alarm_hist_inc fields terminated by "," enclosed by "^" lines terminated by "\r\n";
load data infile "button_authority.csv" into table button_authority fields terminated by "," enclosed by "^" lines terminated by "\r\n";
load data infile "c3_sms_hist_inc.csv" into table c3_sms_hist_inc fields terminated by "," enclosed by "^" lines terminated by "\r\n";
load data infile "datapermisson.csv" into table datapermisson fields terminated by "," enclosed by "^" lines terminated by "\r\n";

CREATE OR REPLACE PROCEDURE P_ETL_ORA_DATA
(
  P_ORA_DIR  VARCHAR2,
  P_DATA_PATH VARCHAR2
) IS
  TYPE T_REC IS RECORD(
    TBN VARCHAR2(40),
    WHR VARCHAR2(4000));
  TYPE T_TABS IS TABLE OF T_REC;
  V_TABS   T_TABS := T_TABS();
  V_ETL_DIR  VARCHAR2(40) := P_ORA_DIR;
  V_LOAD_FILE UTL_FILE.FILE_TYPE;
  PROCEDURE ETL_DATA
  (
    P_SQL_STMT VARCHAR2,
    P_DATA_PATH VARCHAR2,
    P_TB_NAME  VARCHAR2
  ) IS
  BEGIN
    DECLARE
      V_VAR_COL  VARCHAR2(32767);
      V_NUM_COL  NUMBER;
      V_DATE_COL DATE;
      V_TMZ    TIMESTAMP;
      V_COLS   NUMBER;
      V_COLS_DESC DBMS_SQL.DESC_TAB;
      V_ROW_STR  VARCHAR2(32767);
      V_COL_STR  VARCHAR2(32767);
      V_SQL_ID  NUMBER;
      V_SQL_REF  SYS_REFCURSOR;
      V_EXP_FILE UTL_FILE.FILE_TYPE;
      V_DATA_PATH VARCHAR2(200);
    BEGIN
      V_DATA_PATH := P_DATA_PATH;
      IF REGEXP_SUBSTR(V_DATA_PATH, '\\$') IS NULL
      THEN
        V_DATA_PATH := V_DATA_PATH || '\';
      END IF;
      V_DATA_PATH := REPLACE(V_DATA_PATH, '\', '\\');
      OPEN V_SQL_REF FOR P_SQL_STMT;
      V_SQL_ID := DBMS_SQL.TO_CURSOR_NUMBER(V_SQL_REF);
      DBMS_SQL.DESCRIBE_COLUMNS(V_SQL_ID, V_COLS, V_COLS_DESC);
      FOR I IN V_COLS_DESC.FIRST .. V_COLS_DESC.LAST
      LOOP
        CASE
          WHEN V_COLS_DESC(I).COL_TYPE IN (1, 9, 96) THEN
            DBMS_SQL.DEFINE_COLUMN(V_SQL_ID, I, V_VAR_COL, 32767);
          WHEN V_COLS_DESC(I).COL_TYPE = 2 THEN
            DBMS_SQL.DEFINE_COLUMN(V_SQL_ID, I, V_NUM_COL);
          WHEN V_COLS_DESC(I).COL_TYPE = 12 THEN
            DBMS_SQL.DEFINE_COLUMN(V_SQL_ID, I, V_DATE_COL);
          WHEN V_COLS_DESC(I).COL_TYPE = 180 THEN
            DBMS_SQL.DEFINE_COLUMN(V_SQL_ID, I, V_TMZ);
        END CASE;
      END LOOP;
      DECLARE
        V_FLUSH_OVER PLS_INTEGER := 1;
        V_FILE_OVER PLS_INTEGER := 1;
        V_FILE_NO  PLS_INTEGER := 1;
        V_FILE_NAME VARCHAR2(200);
        V_LINE    VARCHAR2(400);
      BEGIN
        WHILE DBMS_SQL.FETCH_ROWS(V_SQL_ID) > 0
        LOOP
          IF V_FILE_OVER = 1
          THEN
            V_FILE_NAME := P_TB_NAME || '_' || V_FILE_NO || '.csv';
            V_EXP_FILE := UTL_FILE.FOPEN(V_ETL_DIR, V_FILE_NAME, OPEN_MODE => 'w', MAX_LINESIZE => 32767);
          END IF;
          V_ROW_STR := '';
          FOR I IN 1 .. V_COLS
          LOOP
            V_COL_STR := '\N';
            BEGIN
              CASE
                WHEN V_COLS_DESC(I).COL_TYPE IN (1, 9, 96) THEN
                  DBMS_SQL.COLUMN_VALUE(V_SQL_ID, I, V_VAR_COL);
                  IF V_VAR_COL IS NOT NULL
                  THEN
                    V_COL_STR := '^' || V_VAR_COL || '^';
                  END IF;
                WHEN V_COLS_DESC(I).COL_TYPE = 2 THEN
                  DBMS_SQL.COLUMN_VALUE(V_SQL_ID, I, V_NUM_COL);
                  IF V_NUM_COL IS NOT NULL
                  THEN
                    V_COL_STR := V_NUM_COL;
                  END IF;
                WHEN V_COLS_DESC(I).COL_TYPE = 12 THEN
                  DBMS_SQL.COLUMN_VALUE(V_SQL_ID, I, V_DATE_COL);
                  IF V_DATE_COL IS NOT NULL
                  THEN
                    V_COL_STR := '^' || TO_CHAR(V_DATE_COL, 'yyyy-mm-dd hh24:mi:ss') || '^';
                  END IF;
                WHEN V_COLS_DESC(I).COL_TYPE IN (180, 181, 231) THEN
                  DBMS_SQL.COLUMN_VALUE(V_SQL_ID, I, V_TMZ);
                  IF V_TMZ IS NOT NULL
                  THEN
                    V_COL_STR := '^' || TO_CHAR(V_TMZ, 'yyyy-mm-dd hh24:mi:ss.ff6') || '^';
                  END IF;
              END CASE;
              IF I = 1
              THEN
                V_ROW_STR := V_COL_STR;
              ELSE
                V_ROW_STR := V_ROW_STR || ',' || V_COL_STR;
              END IF;
            END;
          END LOOP;
          UTL_FILE.PUT_LINE(V_EXP_FILE, CONVERT(V_ROW_STR, 'UTF8'));
          IF V_FILE_OVER > 200000 /*每200000条记录就产生一个新的文件*/
          THEN
            V_FILE_OVER := 1;
            V_FLUSH_OVER := 1;
            V_FILE_NO  := V_FILE_NO + 1;
            UTL_FILE.FCLOSE(V_EXP_FILE);
            V_LINE := 'load data infile "' || V_DATA_PATH || V_FILE_NAME || '" into table ' || P_TB_NAME;
            V_LINE := V_LINE || ' fields terminated by "," enclosed by "^" lines terminated by "\r\n";';
            UTL_FILE.PUT_LINE(V_LOAD_FILE, V_LINE);
            UTL_FILE.FFLUSH(V_LOAD_FILE);
            CONTINUE;
          END IF;
          V_FILE_OVER := V_FILE_OVER + 1;
          IF V_FLUSH_OVER > 2000 /*每2000条记录就刷新缓存，写到文件中 */
          THEN
            UTL_FILE.FFLUSH(V_EXP_FILE);
            V_FLUSH_OVER := 1;
          ELSE
            V_FLUSH_OVER := V_FLUSH_OVER + 1;
          END IF;
        END LOOP;
        DBMS_SQL.CLOSE_CURSOR(V_SQL_ID);
        IF UTL_FILE.IS_OPEN(V_EXP_FILE)
        THEN
          UTL_FILE.FCLOSE(V_EXP_FILE);
          V_LINE := 'load data infile "' || V_DATA_PATH || V_FILE_NAME || '" into table ' || P_TB_NAME;
          V_LINE := V_LINE || ' fields terminated by "," enclosed by "^" lines terminated by "\r\n";';
          UTL_FILE.PUT_LINE(V_LOAD_FILE, V_LINE);
          UTL_FILE.FFLUSH(V_LOAD_FILE);
        END IF;
      END;
    EXCEPTION
      WHEN OTHERS THEN
        IF DBMS_SQL.IS_OPEN(V_SQL_ID)
        THEN
          DBMS_SQL.CLOSE_CURSOR(V_SQL_ID);
        END IF;
        IF UTL_FILE.IS_OPEN(V_EXP_FILE)
        THEN
          UTL_FILE.FCLOSE(V_EXP_FILE);
        END IF;
        DBMS_OUTPUT.PUT_LINE(SQLERRM);
        DBMS_OUTPUT.PUT_LINE(P_SQL_STMT);
    END;
  END;
BEGIN
  BEGIN
    EXECUTE IMMEDIATE 'create table mysql_etl_tbs(tn varchar2(40),cn varchar2(40),ci number) ';
  EXCEPTION
    WHEN OTHERS THEN
      NULL;
  END;
  EXECUTE IMMEDIATE 'truncate table mysql_etl_tbs';
  DECLARE
    V_CI    PLS_INTEGER;
    V_CN    VARCHAR2(40);
    V_ETL_COLS VARCHAR2(32767);
    V_TBN   VARCHAR2(30);
    V_ETL_CFG VARCHAR2(32767);
    V_CNF_FILE UTL_FILE.FILE_TYPE;
    V_FROM_POS PLS_INTEGER;
  BEGIN
    V_CNF_FILE := UTL_FILE.FOPEN(V_ETL_DIR, 'ETL_TABS.CNF', 'r', 32767);
    LOOP
      UTL_FILE.GET_LINE(V_CNF_FILE, V_ETL_CFG, 32767);
      V_FROM_POS := REGEXP_INSTR(V_ETL_CFG, 'from', 1, 1, 0, 'i');
      V_ETL_COLS := SUBSTR(V_ETL_CFG, 1, V_FROM_POS - 1);
      V_ETL_COLS := REGEXP_SUBSTR(V_ETL_COLS, '(select)(.+)', 1, 1, 'i', 2);
      V_TBN   := REGEXP_SUBSTR(V_ETL_CFG, '(\s+from\s+)(\w+)(\s*)', 1, 1, 'i', 2);
      V_TBN   := UPPER(V_TBN);
      V_TABS.EXTEND();
      V_TABS(V_TABS.LAST).TBN := V_TBN;
      V_TABS(V_TABS.LAST).WHR := REGEXP_SUBSTR(V_ETL_CFG, '\s+where .+', 1, 1, 'i');
      V_CI := 1;
      LOOP
        V_CN := REGEXP_SUBSTR(V_ETL_COLS, '\S+', 1, V_CI);
        EXIT WHEN V_CN IS NULL;
        V_CN := UPPER(V_CN);
        EXECUTE IMMEDIATE 'insert into mysql_etl_tbs(tn,cn,ci) values(:1,:2,:3)'
          USING V_TBN, V_CN, V_CI;
        COMMIT;
        V_CI := V_CI + 1;
      END LOOP;
    END LOOP;
  EXCEPTION
    WHEN UTL_FILE.INVALID_PATH THEN
      DBMS_OUTPUT.PUT_LINE('指定的目录：ETL_DIR"' || '"无效！');
      RETURN;
    WHEN UTL_FILE.INVALID_FILENAME THEN
      DBMS_OUTPUT.PUT_LINE('指定的文件：" ETL_TABS.CNF' || '"无效！');
      RETURN;
    WHEN NO_DATA_FOUND THEN
      UTL_FILE.FCLOSE(V_CNF_FILE);
    WHEN OTHERS THEN
      DBMS_OUTPUT.PUT_LINE(SQLERRM);
      RETURN;
  END;
  DECLARE
    V_CUR_MATCH  SYS_REFCURSOR;
    V_SQL_SMT   VARCHAR2(32767);
    V_TN     VARCHAR2(40);
    V_CN     VARCHAR2(40);
    V_CI     PLS_INTEGER;
    V_COLUMN_NAME VARCHAR2(40);
    V_ETL_COLS  VARCHAR2(32767);
    V_LINE    VARCHAR2(4000);
    V_TBN     VARCHAR2(40);
  BEGIN
    V_LOAD_FILE := UTL_FILE.FOPEN(V_ETL_DIR, 'load_data.sql', OPEN_MODE => 'w', MAX_LINESIZE => 32767);
    FOR T_IX IN V_TABS.FIRST .. V_TABS.LAST
    LOOP
      V_SQL_SMT := 'select tn,cn,column_name,ci from ( select * from mysql_etl_tbs where tn='':tbn:'' ) l left join user_tab_columns r on l.tn = r.table_name and l.cn= r.column_name order by ci';
      V_TBN   := V_TABS(T_IX).TBN;
      V_SQL_SMT := REPLACE(V_SQL_SMT, ':tbn:', V_TBN);
      V_ETL_COLS := NULL;
      OPEN V_CUR_MATCH FOR V_SQL_SMT;
      LOOP
        FETCH V_CUR_MATCH
          INTO V_TN, V_CN, V_COLUMN_NAME, V_CI;
        EXIT WHEN V_CUR_MATCH%NOTFOUND;
        IF V_CI > 1
        THEN
          V_ETL_COLS := V_ETL_COLS || ' , ';
        END IF;
        IF V_COLUMN_NAME IS NULL
        THEN
          V_ETL_COLS := V_ETL_COLS || ' cast(null as number) ' || V_CN;
        ELSE
          V_ETL_COLS := V_ETL_COLS || V_CN;
        END IF;
      END LOOP;
      CLOSE V_CUR_MATCH;
      V_TBN   := LOWER(V_TBN);
      V_SQL_SMT := 'select ' || V_ETL_COLS || ' from ' || V_TBN || V_TABS(T_IX).WHR;
      ETL_DATA(V_SQL_SMT, P_DATA_PATH, V_TBN);
    END LOOP;
    IF UTL_FILE.IS_OPEN(V_LOAD_FILE)
    THEN
      UTL_FILE.FCLOSE(V_LOAD_FILE);
    END IF;
  END;
END P_ETL_ORA_DATA;