python_db_merge_check.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#기본 import 함수들\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "from pandas import DataFrame, Timestamp\n",
    "from collections import Counter\n",
    "import os\n",
    "\n",
    "import time\n",
    "from datetime import datetime, timedelta"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def checking_million_count(SRC_sql, SRC_con, counting=1000000):\n",
    "    # count먼저 수행해서 오래 걸리는거만 걸러내기\n",
    "\n",
    "    SRC_count_query = \"SELECT COUNT(*) FROM\"  #맨 앞에만 count를 적용하기 위함\n",
    "    SRC_only_count_use = SRC_sql.split('FROM')\n",
    "    for iii in range(1, len(SRC_only_count_use) - 1):\n",
    "        SRC_count_query = SRC_count_query + SRC_only_count_use[iii] + \"FROM\"\n",
    "    SRC_count_query = SRC_count_query + SRC_only_count_use[-1]\n",
    "\n",
    "    try:\n",
    "        SRC_count_result = pd.read_sql(SRC_count_query, con=SRC_con)\n",
    "        if SRC_count_result.iloc[0][0] >= counting:\n",
    "            result = [\n",
    "                counting, counting, 'SRC_테이블 count ' + str(counting) + '건 이상',\n",
    "                '인코딩 미 확인',\n",
    "                datetime.now()\n",
    "            ]\n",
    "            return result\n",
    "        else:\n",
    "            return None\n",
    "    except:\n",
    "        result = [\n",
    "            -9999, -9999, 'SRC_count_SQL READ ERROR', '인코딩 미 확인',\n",
    "            datetime.now()\n",
    "        ]\n",
    "        return result\n",
    "\n",
    "\n",
    "def comparing_column(SRC_result, TGT_result):\n",
    "    SRC_tm, TGT_tm = SRC_result.columns, TGT_result.columns\n",
    "\n",
    "    if sorted(SRC_tm) != sorted(TGT_tm):\n",
    "        SRC_sub_TGT = [x for x in SRC_tm if x not in TGT_tm]\n",
    "        TGT_sub_SRC = [x for x in TGT_tm if x not in SRC_tm]\n",
    "\n",
    "        SRC1 = ''\n",
    "        for jkl in SRC_sub_TGT:\n",
    "            SRC1 = SRC1 + jkl + ', '\n",
    "        SRC_note = '소스 데이터에 ' + SRC1[:-2] + \" 컬럼이 추가로 있습니다.\"\n",
    "        TGT1 = ''\n",
    "        for jkl in TGT_sub_SRC:\n",
    "            TGT1 = TGT1 + jkl + ', '\n",
    "        TGT_note = '타겟 데이터에 ' + TGT1[:-2] + \" 컬럼이 추가로 있습니다.\"\n",
    "\n",
    "        if SRC1[:-2] == '':\n",
    "            note = TGT_note\n",
    "        elif TGT1[:-2] == '':\n",
    "            note = SRC_note\n",
    "        else:\n",
    "            note = SRC_note + ' ' + TGT_note\n",
    "\n",
    "        result = [-9999, -9999, note, '인코딩 미 확인', datetime.now()]\n",
    "    else:\n",
    "        result = None\n",
    "\n",
    "    return result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def type_check_change(DB_Name, SCHEMA, Table_Name, sql_con, sql_result,\n",
    "                      SRC_tm):\n",
    "\n",
    "    if DB_Name == 'DB2':  #DB2\n",
    "        sql_type_check_query = \"select colname, typename from syscat.columns where tabschema = '\" + SCHEMA + \"' and tabname = '\" + Table_Name + \"'\"\n",
    "        sql_type_check = pd.read_sql(sql_type_check_query, con=sql_con)\n",
    "        sql_type_check = sql_type_check.rename(columns={\n",
    "            \"COLNAME\": \"COLUMN_NAME\",\n",
    "            \"TYPENAME\": \"DATA_TYPE\"\n",
    "        })\n",
    "    elif DB_Name == 'POSTGRE':  #postgre\n",
    "        gp1 = \"select a.attname, c.typname from   pg_catalog.pg_attribute a, pg_catalog.pg_class b, pg_catalog.pg_type c, pg_catalog.pg_tables d\"\n",
    "        gp2 = \" where a.attrelid = b.oid and a.atttypid = c.oid and b.relname  = d.tablename and a.attnum > 0 and d.tablename = '\"\n",
    "        sql_type_check_query = gp1 + gp2 + Table_Name.lower(\n",
    "        ) + \"' and d.schemaname = '\" + SCHEMA + \"'\"\n",
    "        sql_type_check = pd.read_sql(sql_type_check_query, con=sql_con)\n",
    "        sql_type_check = sql_type_check.rename(columns={\n",
    "            \"attname\": \"COLUMN_NAME\",\n",
    "            \"typname\": \"DATA_TYPE\"\n",
    "        })\n",
    "        sql_type_check['COLUMN_NAME'] = sql_type_check[\n",
    "            'COLUMN_NAME'].str.upper()\n",
    "        sql_type_check['DATA_TYPE'] = sql_type_check['DATA_TYPE'].str.upper()\n",
    "    else:  #oracle\n",
    "        sql_type_check_query = \"SELECT column_name, data_type FROM all_tab_columns where table_name =  '\" + Table_Name + \"' AND OWNER = '\" + SCHEMA + \"'\"\n",
    "        sql_type_check = pd.read_sql(sql_type_check_query, con=sql_con)\n",
    "\n",
    "    sql_type_check = sql_type_check[sql_type_check['DATA_TYPE'] != 'UNDEFINED']\n",
    "    sql_type_check = sql_type_check[sql_type_check['COLUMN_NAME'].isin(\n",
    "        SRC_tm)].drop_duplicates(keep='first').reset_index(drop=True)\n",
    "\n",
    "    for i in range(sql_type_check.shape[0]):\n",
    "        try:\n",
    "            if sql_type_check['DATA_TYPE'][i] in [\n",
    "                    'FLOAT', 'SMALLINT', 'NUMBER', 'INTEGER', 'BIGINT',\n",
    "                    'DECIMAL', 'NUMERIC', 'REAL', 'DOUBLE'\n",
    "            ]:\n",
    "                sql_result[sql_type_check['COLUMN_NAME'][i]] = sql_result[\n",
    "                    sql_type_check['COLUMN_NAME'][i]].fillna(-999)\n",
    "                sql_result[sql_type_check['COLUMN_NAME'][i]] = pd.to_numeric(\n",
    "                    sql_result[sql_type_check['COLUMN_NAME'][i]])\n",
    "                sql_result[sql_type_check['COLUMN_NAME'][i]] = [\n",
    "                    -999 if i == None else i\n",
    "                    for i in sql_result[sql_type_check['COLUMN_NAME'][i]]\n",
    "                ]\n",
    "            elif sql_type_check['DATA_TYPE'][i] in ['DATE', 'DATETIME']:\n",
    "                sql_result[sql_type_check['COLUMN_NAME'][i]] = sql_result[\n",
    "                    sql_type_check['COLUMN_NAME'][i]].fillna(\n",
    "                        pd.to_datetime('1990-01-01'))\n",
    "                sql_result[sql_type_check['COLUMN_NAME'][i]] = [\n",
    "                    pd.to_datetime('1990-01-01') if i == None else i\n",
    "                    for i in sql_result[sql_type_check['COLUMN_NAME'][i]]\n",
    "                ]\n",
    "            elif sql_type_check['DATA_TYPE'][i].find('TIMESTAMP') >= 0:\n",
    "                sql_result[sql_type_check['COLUMN_NAME'][i]] = sql_result[\n",
    "                    sql_type_check['COLUMN_NAME'][i]].fillna(\n",
    "                        pd.Timestamp(1990, 1, 1, 0, 0, 0))\n",
    "                sql_result[sql_type_check['COLUMN_NAME'][i]] = [\n",
    "                    pd.Timestamp(1990, 1, 1, 0, 0, 0) if i == None else i\n",
    "                    for i in sql_result[sql_type_check['COLUMN_NAME'][i]]\n",
    "                ]\n",
    "            else:\n",
    "                sql_result[sql_type_check['COLUMN_NAME'][i]] = sql_result[\n",
    "                    sql_type_check['COLUMN_NAME'][i]].fillna(\"대체\").replace(\n",
    "                        '', \"대체\").replace(' ', \"대체\").replace('-', \"대체\")\n",
    "                sql_result[sql_type_check['COLUMN_NAME'][i]] = [\n",
    "                    '대체' if i == None else i\n",
    "                    for i in sql_result[sql_type_check['COLUMN_NAME'][i]]\n",
    "                ]\n",
    "\n",
    "            if sql_type_check['DATA_TYPE'][i].find('TIMESTAMP') >= 0:\n",
    "                sql_result[sql_type_check['COLUMN_NAME'][i]] = [\n",
    "                    i.strftime('%Y-%m-%d-%H:%M:%S')\n",
    "                    for i in sql_result[sql_type_check['COLUMN_NAME'][i]]\n",
    "                ]\n",
    "            elif sql_type_check['DATA_TYPE'][i].find('DATE') >= 0:\n",
    "                sql_result[sql_type_check['COLUMN_NAME'][i]] = [\n",
    "                    i.strftime('%Y-%m-%d-%H:%M:%S')\n",
    "                    for i in sql_result[sql_type_check['COLUMN_NAME'][i]]\n",
    "                ]\n",
    "            else:\n",
    "                pass\n",
    "\n",
    "        except:\n",
    "            pass\n",
    "\n",
    "    return sql_result, sql_type_check"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 반환 받은 SRC, TGT를 merge해서 결과를 노트에 넣는 함수\n",
    "def source_target_merge(SRC_result, TGT_result):\n",
    "    try:\n",
    "        total_result = pd.merge(SRC_result,\n",
    "                                TGT_result,\n",
    "                                how='outer',\n",
    "                                indicator=True)\n",
    "    except:\n",
    "        result = [\n",
    "            -9999, -9999, 'dataframe merge error. please, check data type',\n",
    "            '인코딩 미 확인',\n",
    "            datetime.now()\n",
    "        ]\n",
    "    else:\n",
    "        left = Counter(total_result['_merge'])['left_only']\n",
    "        right = Counter(total_result['_merge'])['right_only']\n",
    "        both = Counter(total_result['_merge'])['both']\n",
    "        t_count = total_result.shape[0]\n",
    "\n",
    "        if ((left == 0) & (right == 0) & (both == 0)) == True:\n",
    "            note = 'SRC, TGT AS-IS 0건'\n",
    "        elif ((left == 0) & (right == 0)) == True:\n",
    "            note = str(t_count) + '건 모두 일치'\n",
    "        elif ((left == 0) & (right != 0)) == True:\n",
    "            note = 'TGT 테이블만 갯수 많음'\n",
    "        elif ((left != 0) & (right == 0)) == True:\n",
    "            note = 'SRC 테이블만 갯수 많음'\n",
    "        else:\n",
    "            note = '데이터 불일치'\n",
    "\n",
    "        result = [left, right, note]\n",
    "\n",
    "    return result, total_result\n",
    "\n",
    "\n",
    "def source_target_merge_oracle_postgre(SRC_result, SRC_type_check,\n",
    "                                       SRC_table_name, TGT_result,\n",
    "                                       TGT_type_check, TGT_table_name):\n",
    "    try:\n",
    "        result, total_result = source_target_merge(SRC_result, TGT_result)\n",
    "        return result, total_result\n",
    "    except:\n",
    "        abc = pd.merge(SRC_type_check, TGT_type_check, on='COLUMN_NAME')\n",
    "        for it in range(abc.shape[0]):\n",
    "            if (abc['DATA_TYPE_x'][it] == 'NUMBER') & (abc['DATA_TYPE_y'][it]\n",
    "                                                       == 'VARCHAR'):\n",
    "                SRC_result[abc['COLUMN_NAME'][it]] = SRC_result[\n",
    "                    abc['COLUMN_NAME'][it]].astype(str)\n",
    "                SRC_result[abc['COLUMN_NAME'][it]] = SRC_result[\n",
    "                    abc['COLUMN_NAME'][it]].replace(\"-999\", \"대체\")\n",
    "            elif (abc['DATA_TYPE_x'][it].find(\"TIMESTAMP\") >=\n",
    "                  0) & (abc['DATA_TYPE_y'][it] == 'DATE'):\n",
    "                SRC_result[abc['COLUMN_NAME'][it]] = SRC_result[\n",
    "                    abc['COLUMN_NAME'][it]].astype(str)\n",
    "                SRC_result[abc['COLUMN_NAME'][it]] = [\n",
    "                    i[0:10] for i in SRC_result[abc['COLUMN_NAME'][it]]\n",
    "                ]\n",
    "                SRC_result[abc['COLUMN_NAME'][it]] = SRC_result[\n",
    "                    abc['COLUMN_NAME'][it]].replace('1990-01-01', \"대체\")\n",
    "                TGT_result[abc['COLUMN_NAME'][it]] = TGT_result[\n",
    "                    abc['COLUMN_NAME'][it]].astype(str)\n",
    "                TGT_result[abc['COLUMN_NAME'][it]] = TGT_result[\n",
    "                    abc['COLUMN_NAME'][it]].replace('1990-01-01', \"대체\")\n",
    "            elif (abc['DATA_TYPE_x'][it]\n",
    "                  == 'VARCHAR2') & (abc['DATA_TYPE_y'][it] == 'NUMERIC'):\n",
    "                try:\n",
    "                    SRC_result[abc['COLUMN_NAME'][it]] = SRC_result[\n",
    "                        abc['COLUMN_NAME'][it]].replace(\"대체\", -999)\n",
    "                    SRC_result[abc['COLUMN_NAME'][it]] = SRC_result[\n",
    "                        abc['COLUMN_NAME'][it]].astype(int)\n",
    "                except:\n",
    "                    pass\n",
    "            else:\n",
    "                pass\n",
    "        try:\n",
    "            result, total_result = source_target_merge(SRC_result, TGT_result)\n",
    "            return result, total_result\n",
    "        except:\n",
    "            result = [\n",
    "                -9999, -9999, 'dataframe merge error. please, check data type',\n",
    "                '인코딩 미 확인',\n",
    "                datetime.now()\n",
    "            ]\n",
    "            total_result = abc\n",
    "            total_result.columns = [\n",
    "                'COLUMN_NAME', 'SRC_COLUMN_TYPE', 'TGT_COLUMN_TYPE'\n",
    "            ]\n",
    "            pass\n",
    "        pass\n",
    "    return result, total_result\n",
    "\n",
    "\n",
    "def checking_encoding(fail_list, result, SRC_result, TGT_result):\n",
    "\n",
    "    SRC_encoding, TGT_encoding = [], []\n",
    "\n",
    "    for col_col in SRC_result.columns:\n",
    "        for jj in fail_list:\n",
    "            try:\n",
    "                if SRC_result[col_col].str.contains(jj).sum() > 0:\n",
    "                    SRC_encoding.append(col_col + jj)\n",
    "            except:\n",
    "                pass\n",
    "    for col_col in TGT_result.columns:\n",
    "        for jj in fail_list:\n",
    "            try:\n",
    "                if TGT_result[col_col].str.contains(jj).sum() > 0:\n",
    "                    TGT_encoding.append(col_col + jj)\n",
    "            except:\n",
    "                pass\n",
    "\n",
    "    if len(SRC_encoding) > 0 and len(TGT_encoding) > 0:\n",
    "        result.append(\"SRC TGT 인코딩 확인 바람\")\n",
    "    elif len(SRC_encoding) > 0 and len(TGT_encoding) == 0:\n",
    "        result.append(\"SRC 인코딩 확인 바람\")\n",
    "    elif len(SRC_encoding) == 0 and len(TGT_encoding) > 0:\n",
    "        result.append(\"TGT 인코딩 확인 바람\")\n",
    "    else:\n",
    "        result.append(\"인코딩 이상 없음\")\n",
    "\n",
    "    result.append(datetime.now())\n",
    "\n",
    "    return result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def total_merge_check(SRC_db_name,\n",
    "                      SRC_SCHEMA,\n",
    "                      SRC_table_name,\n",
    "                      SRC_sql,\n",
    "                      SRC_con,\n",
    "                      TGT_db_name,\n",
    "                      TGT_SCHEMA,\n",
    "                      TGT_table_name,\n",
    "                      TGT_sql,\n",
    "                      TGT_con,\n",
    "                      counter=1000000,\n",
    "                      fail_list=['⑦', ' ', '⑹', 'ㅼ', 'ъ']):\n",
    "\n",
    "    SRC_sql = SRC_sql.replace(';',\n",
    "                              '')  #쿼리 마지막에 있는 ;가 있으면 쿼리 자체가 실행이 안되는 경우가 있어 제거함\n",
    "    TGT_sql = TGT_sql.replace(';', '')\n",
    "    SRC_note, TGT_note = '', ''\n",
    "    result_columns = [\n",
    "        'SRC_not_equal', 'TGT_not_equal', 'result_note', 'table_encoding',\n",
    "        'checking_time'\n",
    "    ]\n",
    "\n",
    "    result = checking_million_count(SRC_sql, SRC_con, counting=counter)\n",
    "\n",
    "    try:\n",
    "        SRC_result = pd.read_sql(SRC_sql, con=SRC_con)\n",
    "        SRC_result.columns = map(lambda x: str(x).upper(), SRC_result.columns)\n",
    "        SRC_result_backup = SRC_result.copy()\n",
    "    except:\n",
    "        SRC_note = \"SRC SQL ERROR\"\n",
    "    try:\n",
    "        TGT_result = pd.read_sql(TGT_sql, con=TGT_con)\n",
    "        TGT_result.columns = map(lambda x: str(x).upper(), TGT_result.columns)\n",
    "        TGT_result_backup = TGT_result.copy()\n",
    "    except:\n",
    "        TGT_note = \"TGT SQL ERROR\"\n",
    "\n",
    "    if SRC_note == \"\" and TGT_note == \"\":\n",
    "        pass\n",
    "    elif SRC_note == \"\" and TGT_note != \"\":\n",
    "        result = [\n",
    "            SRC_result.shape[0], -9999, TGT_note, '인코딩 미 확인',\n",
    "            datetime.now()\n",
    "        ]\n",
    "    elif SRC_note != \"\" and TGT_note == \"\":\n",
    "        result = [\n",
    "            -9999, TGT_result.shape[0], SRC_note, '인코딩 미 확인',\n",
    "            datetime.now()\n",
    "        ]\n",
    "    else:\n",
    "        result = [-9999, -9999, \"ALL SQL ERROR\", '인코딩 미 확인', datetime.now()]\n",
    "\n",
    "    for i in range(1):\n",
    "        if result != None:\n",
    "            return DataFrame([result], columns=result_columns), DataFrame()\n",
    "            continue\n",
    "\n",
    "    SRC_tm, TGT_tm = SRC_result.columns, TGT_result.columns\n",
    "\n",
    "    result = comparing_column(SRC_result, TGT_result)\n",
    "\n",
    "    for i in range(1):\n",
    "        if result != None:\n",
    "            return DataFrame([result], columns=result_columns), DataFrame()\n",
    "            continue\n",
    "\n",
    "    #DB2, Oracle 타입별로 결측치 채우기\n",
    "    SRC_result = SRC_result[SRC_tm]\n",
    "    TGT_result = TGT_result[TGT_tm]\n",
    "\n",
    "    SRC_result, SRC_type_check = type_check_change(SRC_db_name, SRC_SCHEMA,\n",
    "                                                   SRC_table_name, SRC_con,\n",
    "                                                   SRC_result, SRC_tm)\n",
    "    TGT_result, TGT_type_check = type_check_change(TGT_db_name, TGT_SCHEMA,\n",
    "                                                   TGT_table_name, TGT_con,\n",
    "                                                   TGT_result, TGT_tm)\n",
    "\n",
    "    if SRC_db_name == 'ORACLE' and TGT_db_name == 'POSTGRE':\n",
    "        result, total_result = source_target_merge_oracle_postgre(\n",
    "            SRC_result, SRC_type_check, SRC_table_name, TGT_result,\n",
    "            TGT_type_check, TGT_table_name)\n",
    "    else:\n",
    "        result, total_result = source_target_merge(SRC_result, TGT_result)\n",
    "\n",
    "    for i in range(1):\n",
    "        if len(result) == 5:\n",
    "            return DataFrame([result], columns=result_columns), total_result\n",
    "            continue\n",
    "\n",
    "    result = checking_encoding(fail_list, result, SRC_result, TGT_result)\n",
    "\n",
    "    return DataFrame([result],\n",
    "                     columns=result_columns), total_result.sort_values(\n",
    "                         list(total_result.columns))"
   ]
  }
 ],
 "metadata": {
  "hide_input": false,
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}