[2022 공간빅데이터 경진대회] 3-2. 건축물대장 표제부 업로드

이번 글은

• 앞선 [2022 공간빅데이터 경진대회] 3. 데이터 확보 및 DB 구축의 소스 데이터 중 건축물대장 표제부 데이터를 업로드
• 다운받은 소스 데이터 파일(.zip)을 50만줄씩 읽어서 업로드
• 코드 파일 첨부(github)
  • 잠깐 후기 : 항상 python(.py)으로 업로드 작업하다가, 블로그 작성에 좋을 것 같아 python notebook(.ipynb)으로 업로드 작업을 해봤는데, 셀 진행 방식에서는 큰 규모의 loop문을 만드는 것이 불편했다.

테이블 스키마 설정

create table building_dong (
  pnu varchar(19),
  building_pk varchar(33),
  road_address_pk varchar(26),
  register_type varchar(2),
  address varchar,
  road_address varchar,
  building_name varchar,
  dong_name varchar,
  main_sub_type varchar,
  site_area numeric,
  bc_area numeric,
  bc_ratio numeric,
  gf_area numeric,
  far_gf_area numeric,
  fa_ratio numeric,
  structure_code varchar(2),
  structure_name varchar,
  structure_detail varchar,
  use_code varchar(5),
  use_name varchar,
  use_detail varchar,
  sedae_count numeric,
  gagu_count numeric,
  ho_count numeric,
  height numeric,
  floor_count numeric,
  under_floor_count numeric,
  permit_date varchar(8),
  construction_date varchar(8),
  approval_date varchar(8),
  created_date varchar(8),
  uploaded_at timestamp default current_timestamp
);
create index index_building_dong_on_pnu on building_dong (pnu);
create index index_building_dong_on_road_address_pk on building_dong (road_address_pk);

DB 연결

• host = localhost
• db = postgres
• Table Schema = sbd

import psycopg2

conn = psycopg2.connect(
  'host=localhost port=5432 dbname=postgres user=postgres password=postgres',
  options='-c search_path=sbd'
)
conn.set_session(autocommit=True)
cur = conn.cursor()

테이블 생성

cur.execute(
  f'''
  select count(*)
  from information_schema.tables
  where
    table_schema = 'sbd' and
    table_name ~ 'building_dong'
  '''
)
if not cur.fetchone()[0]:
  cur.execute(
    open('sql/source-create_table_building_dong.sql', 'r').read()
  )

후속 작업에 필요한 컬럼 목록, 테이블 초기화

cur.execute(
  f'''
  select column_name
  from information_schema.columns
  where 
    table_schema = 'sbd' and
    table_name = 'building_dong' and
    column_default is null
  '''
)
column_list = [
  col[0]
  for col
  in cur.fetchall()
]

cur.execute(
  'delete from building_dong'
)

업로드 함수

def uploadToDB(data_table):
  print(datetime.now(),'start...',end='')
  data_table.columns = [
    '관리_건축물대장_PK', '대장_구분_코드', '대장_구분_코드_명', '대장_종류_코드', '대장_종류_코드_명', '대지_위치', '도로명_대지_위치', '건물_명', '시군구_코드', '법정동_코드', '대지_구분_코드', '번', '지', '특수지_명', '블록', '로트', '외필지_수', '새주소_도로_코드', '새주소_법정동_코드', '새주소_지상지하_코드', '새주소_본_번', '새주소_부_번', '동_명', '주_부속_구분_코드', '주_부속_구분_코드_명', '대지_면적(㎡)', '건축_면적(㎡)', '건폐_율(%)', '연면적(㎡)', '용적_률_산정_연면적(㎡)', '용적_률(%)', '구조_코드', '구조_코드_명', '기타_구조', '주_용도_코드', '주_용도_코드_명', '기타_용도', '지붕_코드', '지붕_코드_명', '기타_지붕', '세대_수(세대)', '가구_수(가구)', '높이(m)', '지상_층_수', '지하_층_수', '승용_승강기_수', '비상용_승강기_수', '부속_건축물_수', '부속_건축물_면적(㎡)', '총_동_연면적(㎡)', '옥내_기계식_대수(대)', '옥내_기계식_면적(㎡)', '옥외_기계식_대수(대)', '옥외_기계식_면적(㎡)', '옥내_자주식_대수(대)', '옥내_자주식_면적(㎡)', '옥외_자주식_대수(대)', '옥외_자주식_면적(㎡)', '허가_일', '착공_일', '사용승인_일', '허가번호_년', '허가번호_기관_코드', '허가번호_기관_코드_명', '허가번호_구분_코드', '허가번호_구분_코드_명', '호_수(호)', '에너지효율_등급', '에너지절감_율', '에너지_EPI점수', '친환경_건축물_등급', '친환경_건축물_인증점수', '지능형_건축물_등급', '지능형_건축물_인증점수', '생성_일자', '내진_설계_적용_여부', '내진_능력'
  ]
  data_table['시도_코드'] = [code[0:2] for code in data_table['시군구_코드']]
  data_table = data_table[data_table['시도_코드'] == '11'] # '전국' 중 '서울'만 추출
  data_table['pnu'] = data_table['시군구_코드'] + data_table['법정동_코드'] + [ # 토지 ID인 PNU 생성
    '1' if code == '0' else '2' if code == '1' else '0'
    for code
    in data_table['대지_구분_코드']
  ] + data_table['번'] + data_table['지']
  data_table['road_address_pk'] = [
    cd1[0:5]+cd2[0:3]+cd1[5:12]+cd3+cd4.zfill(5)+cd5.zfill(5)
      if not pd.isna(cd1) and not pd.isna(cd2) and not pd.isna(cd4) and not pd.isna(cd5)
      else None
    for cd1, cd2, cd3, cd4, cd5
    in data_table[['새주소_도로_코드', '새주소_법정동_코드', '새주소_지상지하_코드', '새주소_본_번', '새주소_부_번']].to_numpy()
  ]
  data_table = data_table[[
    'pnu', '관리_건축물대장_PK', 'road_address_pk', '대장_구분_코드_명', '대지_위치', '도로명_대지_위치', '건물_명', '동_명', '주_부속_구분_코드_명', '대지_면적(㎡)', '건축_면적(㎡)', '건폐_율(%)', '연면적(㎡)', '용적_률_산정_연면적(㎡)', '용적_률(%)', '구조_코드', '구조_코드_명', '기타_구조', '주_용도_코드', '주_용도_코드_명', '기타_용도', '세대_수(세대)', '가구_수(가구)', '호_수(호)', '높이(m)', '지상_층_수', '지하_층_수', '허가_일', '착공_일', '사용승인_일', '생성_일자'
  ]]
  data_table.columns = column_list
  data_table.structure_detail = data_table.structure_detail.replace('[\\\]', '', regex=True) # 전처리 : 특수문자 제거
  data_table.use_detail = data_table.use_detail.replace('[\\\]', '', regex=True) # 전처리 : 특수문자 제거
  data_table.to_csv( # .txt 파일로 로컬 저장
    'temp_building_dong.txt',
    sep='|',
    index=False,
    header=False,
    encoding='CP949'
  )
  temp_file = open('temp_building_dong.txt', 'r')
  print('upload...',end='')
  cur.copy_from( # file bulk insert
    temp_file,
    'building_dong',
    sep='|',
    columns=column_list,
    null=''
  )
  temp_file.close()
  os.remove('temp_building_dong.txt')
  print('end',datetime.now())

소스 데이터 업로드

from zipfile import ZipFile
import pandas as pd
import os
from datetime import datetime

zf = ZipFile('D:/data/building_register/building_dong/국토교통부_건축물대장_표제부+(2022년+07월).zip')
source_chunks = pd.read_csv(
  zf.open('mart_djy_03.txt'),
  sep='|',
  encoding='CP949',
  header=None,
  dtype='string',
  chunksize=500_000
)

for source_dt in source_chunks:
  uploadToDB(source_dt)

• 업로드 데이터 수 : 595,336 rows