Der Implementierungsprozess zum Extrahieren von Oracle-Daten in eine MySQL-Datenbank

Laden Sie die Daten in der Datei „alarm_hist_inc.csv“ in die Tabelle alarm_hist_inc. Felder, die mit „“, abgeschlossen sind, umschlossen von „^“, Zeilen, die mit „\r\n“ abgeschlossen sind.
Lade die Daten in der Datei „button_authority.csv“ in die Tabelle button_authority. Die Felder sind mit „“, „umschlossen von“^“, die Zeilen sind mit „\r\n“ abgeschlossen.
Lade die Daten aus der Datei „c3_sms_hist_inc.csv“ in die Tabelle c3_sms_hist_inc. Die Felder sind mit „“, „umschlossen von“^“, die Zeilen sind mit „\r\n“ abgeschlossen.
Laden Sie die Daten in der Datei „datapermisson.csv“ in die Tabelle „datapermisson“. Die Felder sind mit „“, „umschlossen von“^“, die Zeilen sind mit „\r\n“ abgeschlossen.

ERSTELLEN ODER ERSETZEN DER VERFAHREN P_ETL_ORA_DATA
(
  P_ORA_DIR VARCHAR2,
  P_DATA_PATH VARCHAR2
) IST
  TYP T_REC IST AUFZEICHNUNG(
    TBN VARCHAR2(40),
    WHR VARCHAR2(4000));
  TYP T_TABS IST TABELLE VON T_REC;
  V_TABS T_TABS := T_TABS();
  V_ETL_DIR VARCHAR2(40) := P_ORA_DIR;
  V_LOAD_FILE UTL_FILE.DATEITYP;
  VERFAHREN ETL_DATA
  (
    P_SQL_STMT VARCHAR2,
    P_DATA_PATH VARCHAR2,
    P_TB_NAME VARCHAR2
  ) IST
  BEGINNEN
    ERKLÄREN
      V_VAR_COL VARCHAR2(32767);
      V_NUM_COL-NUMMER;
      V_DATE_COL DATUM;
      V_TMZ-ZEITSTEMPEL;
      V_SPALTENNUMMER;
      V_COLS_DESC DBMS_SQL.DESC_TAB;
      V_ROW_STR VARCHAR2(32767);
      V_COL_STR VARCHAR2(32767);
      V_SQL_ID-NUMMER;
      V_SQL_REF SYS_REFCURSOR;
      V_EXP_FILE UTL_FILE.DATEITYP;
      V_DATA_PATH VARCHAR2(200);
    BEGINNEN
      V_DATA_PATH := P_DATA_PATH;
      WENN REGEXP_SUBSTR(V_DATA_PATH, '\\$') NULL IST
      DANN
        V_DATA_PATH := V_DATA_PATH || '\';
      ENDE, WENN;
      V_DATA_PATH := ERSETZEN(V_DATA_PATH, '\', '\\');
      ÖFFNEN SIE V_SQL_REF FÜR 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);
      FÜR I IN V_COLS_DESC.FIRST .. V_COLS_DESC.LAST
      SCHLEIFE
        FALL
          WENN V_COLS_DESC(I).COL_TYPE IN (1, 9, 96) DANN
            DBMS_SQL.DEFINE_COLUMN(V_SQL_ID, I, V_VAR_COL, 32767);
          WENN V_COLS_DESC(I).COL_TYPE = 2, DANN
            DBMS_SQL.DEFINE_COLUMN(V_SQL_ID, I, V_NUM_COL);
          WENN V_COLS_DESC(I).COL_TYPE = 12, DANN
            DBMS_SQL.DEFINE_COLUMN(V_SQL_ID, I, V_DATE_COL);
          WENN V_COLS_DESC(I).COL_TYPE = 180, DANN
            DBMS_SQL.DEFINE_COLUMN(V_SQL_ID, I, V_TMZ);
        FALLENDE;
      ENDE DER SCHLEIFE;
      ERKLÄREN
        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);
      BEGINNEN
        WHILE DBMS_SQL.FETCH_ROWS(V_SQL_ID) > 0
        SCHLEIFE
          WENN V_FILE_OVER = 1
          DANN
            V_DATEINAME := P_TB_NAME || '_' || V_DATEI_NR || '.csv';
            V_EXP_FILE := UTL_FILE.FOPEN(V_ETL_DIR, V_FILE_NAME, OPEN_MODE => 'w', MAX_LINESIZE => 32767);
          ENDE, WENN;
          V_ROW_STR := '';
          FÜR I IN 1 .. V_COLS
          SCHLEIFE
            V_COL_STR := '\N';
            BEGINNEN
              FALL
                WENN V_COLS_DESC(I).COL_TYPE IN (1, 9, 96) DANN
                  DBMS_SQL.COLUMN_VALUE(V_SQL_ID, I, V_VAR_COL);
                  WENN V_VAR_COL NICHT NULL IST
                  DANN
                    V_COL_STR := '^' || V_VAR_COL || '^';
                  ENDE, WENN;
                WENN V_COLS_DESC(I).COL_TYPE = 2, DANN
                  DBMS_SQL.COLUMN_VALUE(V_SQL_ID, I, V_NUM_COL);
                  WENN V_NUM_COL NICHT NULL IST
                  DANN
                    V_COL_STR := V_NUM_COL;
                  ENDE, WENN;
                WENN V_COLS_DESC(I).COL_TYPE = 12, DANN
                  DBMS_SQL.COLUMN_VALUE(V_SQL_ID, I, V_DATE_COL);
                  WENN V_DATE_COL NICHT NULL IST
                  DANN
                    V_COL_STR := '^' || TO_CHAR(V_DATE_COL, 'yyyy-mm-dd hh24:mi:ss') || '^';
                  ENDE, WENN;
                WENN V_COLS_DESC(I).COL_TYPE IN (180, 181, 231) DANN
                  DBMS_SQL.COLUMN_VALUE(V_SQL_ID, I, V_TMZ);
                  WENN V_TMZ NICHT NULL IST
                  DANN
                    V_COL_STR := '^' || TO_CHAR(V_TMZ, 'yyyy-mm-dd hh24:mi:ss.ff6') || '^';
                  ENDE, WENN;
              FALLENDE;
              WENN I = 1
              DANN
                V_ROW_STR := V_COL_STR;
              ANDERS
                V_ROW_STR := V_ROW_STR || ',' || V_COL_STR;
              ENDE, WENN;
            ENDE;
          ENDE DER SCHLEIFE;
          UTL_FILE.PUT_LINE(V_EXP_FILE, KONVERTIEREN(V_ROW_STR, 'UTF8'));
          IF V_FILE_OVER > 200000 /*Erstelle für jeweils 200000 Datensätze eine neue Datei*/
          DANN
            V_FILE_OVER := 1;
            V_FLUSH_OVER := 1;
            V_DATEI_NR := V_DATEI_NR + 1;
            UTL_FILE.FCLOSE(V_EXP_FILE);
            V_LINE := 'Daten in Datei "' || V_DATA_PATH || V_FILE_NAME || '" in Tabelle ' laden || P_TB_NAME;
            V_LINE := V_LINE || ' Felder abgeschlossen durch ",", umschlossen von "^", Zeilen abgeschlossen durch "\r\n";';
            UTL_FILE.PUT_LINE(V_LOAD_FILE, V_LINE);
            UTL_FILE.FFLUSH(V_LOAD_FILE);
            WEITERMACHEN;
          ENDE, WENN;
          V_FILE_OVER := V_FILE_OVER + 1;
          IF V_FLUSH_OVER > 2000 /*Cache aktualisieren und alle 2000 Datensätze in die Datei schreiben*/
          DANN
            UTL_FILE.FFLUSH(V_EXP_FILE);
            V_FLUSH_OVER := 1;
          ANDERS
            V_FLUSH_OVER := V_FLUSH_OVER + 1;
          ENDE, WENN;
        ENDE DER SCHLEIFE;
        DBMS_SQL.CLOSE_CURSOR(V_SQL_ID);
        WENN UTL_FILE.IS_OPEN(V_EXP_FILE)
        DANN
          UTL_FILE.FCLOSE(V_EXP_FILE);
          V_LINE := 'Daten in Datei "' || V_DATA_PATH || V_FILE_NAME || '" in Tabelle ' laden || P_TB_NAME;
          V_LINE := V_LINE || ' Felder abgeschlossen durch ",", umschlossen von "^", Zeilen abgeschlossen durch "\r\n";';
          UTL_FILE.PUT_LINE(V_LOAD_FILE, V_LINE);
          UTL_FILE.FFLUSH(V_LOAD_FILE);
        ENDE, WENN;
      ENDE;
    AUSNAHME
      WENN ANDERE DANN
        WENN DBMS_SQL.IS_OPEN(V_SQL_ID)
        DANN
          DBMS_SQL.CLOSE_CURSOR(V_SQL_ID);
        ENDE, WENN;
        WENN UTL_FILE.IS_OPEN(V_EXP_FILE)
        DANN
          UTL_FILE.FCLOSE(V_EXP_FILE);
        ENDE, WENN;
        DBMS_OUTPUT.PUT_LINE(SQLERRM);
        DBMS_OUTPUT.PUT_LINE(P_SQL_STMT);
    ENDE;
  ENDE;
BEGINNEN
  BEGINNEN
    SOFORT AUSFÜHREN 'Tabelle erstellen mysql_etl_tbs(tn varchar2(40),cn varchar2(40),ci-Nummer)';
  AUSNAHME
    WENN ANDERE DANN
      NULL;
  ENDE;
  SOFORT AUSFÜHREN 'Tabelle mysql_etl_tbs abschneiden';
  ERKLÄREN
    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.DATEITYP;
    V_FROM_POS PLS_INTEGER;
  BEGINNEN
    V_CNF_FILE := UTL_FILE.FOPEN(V_ETL_DIR, 'ETL_TABS.CNF', 'r', 32767);
    SCHLEIFE
      UTL_FILE.GET_LINE(V_CNF_FILE, V_ETL_CFG, 32767);
      V_FROM_POS := REGEXP_INSTR(V_ETL_CFG, 'von', 1, 1, 0, 'i');
      V_ETL_COLS := SUBSTR(V_ETL_CFG, 1, V_FROM_POS - 1);
      V_ETL_COLS := REGEXP_SUBSTR(V_ETL_COLS, '(Auswahl)(.+)', 1, 1, 'i', 2);
      V_TBN := REGEXP_SUBSTR(V_ETL_CFG, '(\s+von\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+wobei .+', 1, 1, 'i');
      V_CI := 1;
      SCHLEIFE
        V_CN := REGEXP_SUBSTR(V_ETL_COLS, '\S+', 1, V_CI);
        BEENDEN, WENN V_CN NULL IST;
        V_CN := UPPER(V_CN);
        SOFORT AUSFÜHREN 'in mysql_etl_tbs(tn,cn,ci) Werte(:1,:2,:3) einfügen'
          VERWENDUNG VON V_TBN, V_CN, V_CI;
        BEGEHEN;
        V_CI := V_CI + 1;
      ENDE DER SCHLEIFE;
    ENDE DER SCHLEIFE;
  AUSNAHME
    WANN UTL_FILE.INVALID_PATH DANN
      DBMS_OUTPUT.PUT_LINE('Angegebenes Verzeichnis: ETL_DIR"' || '"Ungültig!');
      ZURÜCKKEHREN;
    WANN UTL_FILE.INVALID_FILENAME DANN
      DBMS_OUTPUT.PUT_LINE('Angegebene Datei: "ETL_TABS.CNF' || '"Ungültig!');
      ZURÜCKKEHREN;
    WENN KEINE DATEN GEFUNDEN WERDEN, DANN
      UTL_FILE.FCLOSE(V_CNF_FILE);
    WENN ANDERE DANN
      DBMS_OUTPUT.PUT_LINE(SQLERRM);
      ZURÜCKKEHREN;
  ENDE;
  ERKLÄREN
    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);
  BEGINNEN
    V_LOAD_FILE := UTL_FILE.FOPEN(V_ETL_DIR, 'load_data.sql', OPEN_MODE => 'w', MAX_LINESIZE => 32767);
    FÜR T_IX IN V_TABS.FIRST .. V_TABS.LAST
    SCHLEIFE
      V_SQL_SMT := 'Wählen Sie tn, cn, Spaltenname, ci aus (Wählen Sie * aus mysql_etl_tbs, wobei tn = '':tbn:'') l links verbinden Sie user_tab_columns r auf l.tn = r.table_name und l.cn = r.spaltenname, sortiert nach ci';
      V_TBN := V_TABS(T_IX).TBN;
      V_SQL_SMT := ERSETZEN(V_SQL_SMT, ':tbn:', V_TBN);
      V_ETL_COLS := NULL;
      ÖFFNEN SIE V_CUR_MATCH FÜR V_SQL_SMT;
      SCHLEIFE
        FETCH V_CUR_MATCH
          IN V_TN, V_CN, V_COLUMN_NAME, V_CI;
        BEENDEN, WENN V_CUR_MATCH%NICHT GEFUNDEN WIRD;
        WENN V_CI > 1
        DANN
          V_ETL_COLS := V_ETL_COLS || ' , ';
        ENDE, WENN;
        WENN V_COLUMN_NAME NULL IST
        DANN
          V_ETL_COLS := V_ETL_COLS || ' cast(null als Zahl) ' || V_CN;
        ANDERS
          V_ETL_COLS := V_ETL_COLS ||
        ENDE, WENN;
      ENDE DER SCHLEIFE;
      V_CUR_MATCH SCHLIESSEN;
      V_TBN := LOWER(V_TBN);
      V_SQL_SMT := 'Wählen Sie ' || V_ETL_COLS || ' von ' || V_TBN || V_TABS(T_IX).WHR;
      ETL_DATA(V_SQL_SMT, P_DATA_PATH, V_TBN);
    ENDE DER SCHLEIFE;
    WENN UTL_FILE.IS_OPEN(V_LOAD_FILE)
    DANN
      UTL_FILE.FCLOSE(V_LOAD_FILE);
    ENDE, WENN;
  ENDE;
ENDE P_ETL_ORA_DATA;