Upload e integridade dos dados originais

A seguir passo a passo didático que deu origem ao algoritmo "automatizador de ingestão".

Carga para o servidor

Os arquivos precisam ser distribuídos, via FTP ou SFTP, por pastas do município. Por exemplo, com ls /home/igor/ obtemos a listagem das pastas organizadas pelo Igor em junho de 2020:

BR-ES-Serra          BR-PR-Cascavel        BR-RS-PortoAlegre  BR-SP-LaranjalPaulista  BR-SP-SaoVicente
BR-ES-VilaVelha      BR-PR-Pinhais         BR-RS-SantaMaria   BR-SP-Osasco            BR-SP-Sorocaba
BR-ES-Vitoria        BR-PR-SaoJosePinhais  BR-SC-JaraguaSul   BR-SP-Santos
BR-MG-BeloHorizonte  BR-RJ-Niteroi         BR-SP-Itu          BR-SP-SaoBernardoCampo
BR-PE-Recife         BR-RS-Gravatai        BR-SP-Jundiai      BR-SP-SaoPaulo

Dentro de cada uma delas são mantidos os arquivos de input. Por exemplo:

ls /home/igor/BR-RS-PortoAlegre/input
# SMF-XLSX-ORIGINAIS.zip

Avaliação dos arquivos carregados:

Os arquivos de carga podem ser verificados diretamente pelo PostgreSQL, conforme a função ingest.cityfolder_input_files definida abaixo

CREATE or replace FUNCTION ingest.cityfolder_input_files(
  p_fpath text DEFAULT '/tmp/pg_io/'
) RETURNS TABLE (fid int, cityname text, fname text, is_validext boolean, fmeta jsonb) AS $f$
  WITH t0 AS ( SELECT rtrim(p_fpath,'/') AS fpath )
  , t1 AS (
    SELECT f as cityname,
           t0.fpath ||'/'|| f as f
    FROM pg_ls_dir((SELECT fpath FROM t0)) t(f), t0
    WHERE    f ~ '^BR\-[A-Z]{2,2}\-[A-Za-z]+$'
  )
  SELECT (row_number() OVER ())::int id,
         cityname, fname,
         fname ~* '\.(zip|gz|rar|geojson|csv|dwg)$' as is_validext,
         to_jsonb( pg_stat_file(fpath||'/'||fname) ) || jsonb_build_object('fpath',fpath)
  FROM (
    SELECT cityname, 
           f||'/'||'input' as fpath,
           pg_ls_dir(f||'/'||'input') as fname
    FROM t1
    ORDER BY 1,3
  ) t2
$f$  LANGUAGE SQL IMMUTABLE;
-- Exemplo:
SELECT cityname, fname, is_validext, fmeta->>'size' as bytes 
FROM ingest.cityfolder_input_files('/home/igor');

No exemplo, rodando em 1 de julho de 2020, foram listados 243 arquivos válidos:

cityname	fname	is_validext	bytes
BR-ES-Serra	2018-01-05 BaseCartográfica.gdb.zip	t	24218623
BR-ES-VilaVelha	BAIRROS.zip	t	68069
BR-ES-VilaVelha	LOTES.zip	t	6282461
BR-ES-VilaVelha	QUADRAS.zip	t	2327148
...	...	...	...
BR-SP-SaoPaulo	SIRGAS_SHP_quadraviariaed.zip	t	70291605
BR-SP-SaoPaulo	SIRGAS_SHP_subprefeitura.zip	t	1188797
BR-SP-SaoVicente	MAPA DE LOTEAMENTOS.dwg	t	22665728
BR-SP-Sorocaba	lotes_prediais.zip	t	14476669
BR-SP-Sorocaba	shape_eixos_sorocaba.zip	t	1880173

RESUMO DOS ARQUIVOS VÁLIDOS POR MUNICÍPIO:

SELECT cityname, COUNT(*) as n_files,
       sum(bytes/1048576) as "tot MiB",
       ROUND(PERCENTILE_CONT(0.5) WITHIN GROUP(ORDER BY bytes) /1048576 ) as "median MiB"
FROM (
  SELECT cityname, (fmeta->'size')::int as bytes 
  FROM ingest.cityfolder_input_files('/home/igor')
  WHERE is_validExt
) t
GROUP BY 1;

cityname	n_files	tot MiB	median MiB
BR-ES-Serra	1	23	23
BR-ES-VilaVelha	4	10	3
BR-ES-Vitoria	4	16	4
BR-MG-BeloHorizonte	1	40	41
BR-PE-Recife	1	26	27
BR-PR-Cascavel	3	10	2
...	..	..	..
BR-SP-SaoBernardoCampo	5	38	3
BR-SP-SaoPaulo	197	1013	3
BR-SP-SaoVicente	1	21	22
BR-SP-Sorocaba	2	14	8

GERANDO LINHAS DE COMANDO PARA RENOMEAR:

SELECT 'sha256sum "'|| f ||'"'
FROM (
  SELECT fmeta->>'fpath' as fpath, fmeta->>'fpath' ||'/'|| fname as f
  FROM ingest.cityfolder_input_files('/home/igor')
  WHERE is_validExt
) t
;

Amostra dos resultados, e execução como comandos e respectivos resultados:

sha256sum "/home/igor/BR-ES-Serra/input/2018-01-05 BaseCartográfica.gdb.zip"
# b18fc8ebe8bccc2cfdbbbd5f4896d5f6573033ebfa80061d1d86550c5ae8521d  /home/igor/BR-ES-Serra/input/2018-01-05 BaseCartográfica.gdb.zip
sha256sum "/home/igor/BR-ES-VilaVelha/input/BAIRROS.zip"
# c0cd7b2a4cc67b5d49a4d296f41b564b23464364ab746adc6d2206d5dd9249af  /home/igor/BR-ES-VilaVelha/input/BAIRROS.zip

Pode-se portanto criar uma função shell que já execute também o mv, por exemplo

mv  "/home/igor/BR-ES-Serra/input/2018-01-05 BaseCartográfica.gdb.zip" b18fc8ebe8bccc2cfdbbbd5f4896d5f6573033ebfa80061d1d86550c5ae8521d.zip
mv  "/home/igor/BR-ES-VilaVelha/input/BAIRROS.zip" c0cd7b2a4cc67b5d49a4d296f41b564b23464364ab746adc6d2206d5dd9249af.zip

Convenções de nomes de pasta e subpasta e tipos

A cada município a pasta {municipio}/input designa pode conter ou arquivos padrão ou subpastas de arquivos mais especificos.

Os tipos respeitam a sintaxe ... E cada um dos tipos ("content type" ou simplesmente ctype) e scripts de tratamento são descritos na planilha... Para destacar esses tipos a função então foi redefinida como se segue:

CREATE or replace FUNCTION ingest.cityfolder_input_files(
  p_fpath text DEFAULT '/tmp/pg_io/'
) RETURNS TABLE (fid int, cityname text, fname text, ctype text, is_valid boolean, fmeta jsonb) AS $f$

  WITH t0 AS ( SELECT rtrim(p_fpath,'/') AS fpath )
  , t1 AS (
    SELECT f as cityname,
           t0.fpath ||'/'|| f as f
    FROM pg_ls_dir((SELECT fpath FROM t0)) t(f), t0
    WHERE    f ~ '^BR\-[A-Z]{2,2}\-[A-Za-z]+$'
  )
  ,tres AS (
    SELECT cityname, fname, 'std' AS ctype,
         to_jsonb( pg_stat_file(fpath||'/'||fname) ) || jsonb_build_object('fpath',fpath) fmeta 
    FROM (  -- t2:
      SELECT cityname, 
             f||'/'||'input' as fpath,
             pg_ls_dir(f||'/'||'input') as fname
      FROM t1
      ORDER BY 1,3
    ) t2
  ), tres2 AS ( -- main query:

  SELECT (row_number() OVER ())::int id,  
          cityname , fname, ctype,
          fname ~* '\.(zip|gz|rar|geojson|csv|dwg)$' as is_validext,
          fmeta
  FROM ( -- t3:

    SELECT * FROM tres WHERE not((fmeta->'isdir')::boolean)

    UNION 

    SELECT cityname, fname,ctype,
           to_jsonb( pg_stat_file(fpath||'/'||fname) ) || jsonb_build_object('fpath',fpath) AS fmeta 
    FROM ( -- t4:
      select cityname, fname as ctype,
             (fmeta->>'fpath')||'/'|| fname AS fpath,
             pg_ls_dir((fmeta->>'fpath')||'/'||fname) AS fname
      from tres
      where (fmeta->'isdir')::boolean
    ) t4

  ) t3
  ) -- \tres2
  SELECT id, cityname, fname, ctype, is_validext,
         fmeta || CASE WHEN is_validext THEN '{}'::jsonb ELSE jsonb_build_object('is_valid_err','#ER01: file extension unknowed; ') END
  FROM tres2
$f$  LANGUAGE SQL IMMUTABLE;
-- TESTE:
select cityname, ctype, count(*) n, sum(is_valid::int) as n_valid from ingest.cityfolder_input_files('/home/igor') group by 1,2 order by 1,2;

cityname	ctype	n	n_valid
BR-ES-Serra	std	1	1
BR-ES-VilaVelha	std	4	4
...	...	...	...
BR-SP-SaoPaulo	edificacoes	97	96
BR-SP-SaoPaulo	eixos	1	1
BR-SP-SaoPaulo	lotes	97	96
BR-SP-SaoPaulo	planilhas	2	1
BR-SP-SaoPaulo	quadras	1	1
BR-SP-SaoPaulo	std	1	0
BR-SP-SaoPaulo	subdivisao	2	2
BR-SP-SaoVicente	std	1	1
BR-SP-Sorocaba	std	2	2

Join com sha258 da origem

A execussão dos comandos de sha256sum pode ficar a cardo de um script independente, rodando via proprietário ou sudor.

select distinct concat('cd ', fmeta->>'fpath', '; sha256sum -b *.* > sha256sum.txt; chmod 666 sha256sum.txt') as cmd 
from ingest.cityfolder_input_files('/home/igor') order by 1;

Resulta em

# ...
 cd /home/igor/BR-SP-SaoPaulo/input/edificacoes; sha256sum -b *.* > sha256sum.txt; chmod 666 sha256sum.txt
 cd /home/igor/BR-SP-SaoPaulo/input/eixos; sha256sum -b *.* > sha256sum.txt; chmod 666 sha256sum.txt
 cd /home/igor/BR-SP-SaoPaulo/input/lotes; sha256sum -b *.* > sha256sum.txt; chmod 666 sha256sum.txt
 cd /home/igor/BR-SP-SaoPaulo/input/planilhas; sha256sum -b *.* > sha256sum.txt; chmod 666 sha256sum.txt
 cd /home/igor/BR-SP-SaoPaulo/input/quadras; sha256sum -b *.* > sha256sum.txt; chmod 666 sha256sum.txt
 cd /home/igor/BR-SP-SaoPaulo/input; sha256sum -b *.* > sha256sum.txt; chmod 666 sha256sum.txt
 cd /home/igor/BR-SP-SaoPaulo/input/subdivisao; sha256sum -b *.* > sha256sum.txt; chmod 666 sha256sum.txt
# ...

A recuperação dos dados de sha256sum.txt de cada pasta requer uma função de leitura, para o join, uma função final com a coluna do sha256sum de cada arquivo.

CREATE or replace FUNCTION pg_read_file(text,boolean) RETURNS text AS $wrap$
  SELECT pg_read_file($1,0,922337203685477580,$2)
$wrap$ LANGUAGE SQL IMMUTABLE;

CREATE or replace FUNCTION ingest.read_hashsum(
  p_file text -- for example '/tmp/pg_io/sha256sum.txt'
) RETURNS TABLE (hash text, hashtype text, file text, refpath text) AS $f$
  SELECT x[1] as hash,  hashtype, x[2] as file, 
         regexp_replace(p_file, '/?([^/]+)\.txt$', '') -- refpath
  FROM (
    SELECT regexp_split_to_array(line,'\s+\*?') AS x, 
           (regexp_match(p_file, '([^/]+)\.txt$'))[1]
           || '-'
           || CASE WHEN (regexp_match(line, '\s+(\*)'))[1]='*' THEN 'bin' ELSE 'text' END 
           AS hashtype 
    FROM regexp_split_to_table(  pg_read_file(p_file,true),  E'\n'  ) t(line)
  ) t2
  WHERE x is not null AND x[1]>''
$f$  LANGUAGE SQL IMMUTABLE;
-- SELECT hash, file from ingest.read_hashsum('/home/igor/BR-SP-SaoPaulo/input/lotes/sha256sum.txt');

CREATE or replace FUNCTION ingest.cityfolder_input(
  p_fpath text DEFAULT '/tmp/pg_io/',
  checksum_file text DEFAULT 'sha256sum.txt'
) RETURNS TABLE (fid int, cityname text, fname text, ctype text, is_valid boolean, fmeta jsonb) AS $f$
   WITH t AS (
      SELECT *, fmeta->>'fpath' AS fpath FROM ingest.cityfolder_input_files(p_fpath)
   ) 
   SELECT t.fid, t.cityname, t.fname, t.ctype,
          t.is_valid AND k2.hash is not null AS is_valid,
          CASE WHEN k2.hash is not null THEN  t.fmeta || jsonb_build_object('hash',k2.hash, 'hashtype', k2.hashtype)
               ELSE t.fmeta || jsonb_build_object('is_valid_err', COALESCE(t.fmeta->>'is_valid_err','')||'#ER02: hash not generated; ')  END
   FROM t LEFT JOIN (
           SELECT k.*
           FROM (SELECT DISTINCT fmeta->>'fpath' as fpath FROM t) t2, 
                LATERAL ingest.read_hashsum( t2.fpath||'/'||checksum_file) k
           WHERE t2.fpath=k.refpath -- and k.refpath is not null
   ) k2
   ON k2.refpath = t.fpath AND t.fname=k2.file
   WHERE t.fname!=checksum_file -- nome reservado!
   ORDER BY t.cityname, t.fname
$f$  LANGUAGE SQL IMMUTABLE;
/* -- TESTE:
select cityname, ctype, count(*) n, sum(is_valid::int) as n_valid 
from ingest.cityfolder_input('/home/igor') group by 1,2;

select cityname, ctype, count(*) n_nonvalid, array_agg(DISTINCT  fmeta->>'is_valid_err') as errs
from ingest.cityfolder_input('/home/igor') where not(is_valid) group by 1,2;
*/

Exemplo de captura dos hashes gerados pela função ingest.read_hashsum() antes de renomear os arquivos, todos hashtype sha256sum-bin:

hash	file
8d57b8e89e77e7d2c194b7afc208ef861c85e0e233b52eae7dfaaa159ecd7f6a	SIRGAS_SHP_LOTES_01_AGUA_RASA.zip
028b1500b346891bdf2acb04d63f15db4604c5d9f575e4d69ad7e3b1c84d6c74	SIRGAS_SHP_LOTES_02_ALTO_DE_PINHEIROS.zip
96c274bdee92c32dffec56115b63e4c74ce27891b9082033f6c30a1e65491032	SIRGAS_SHP_LOTES_03_ANHANGUERA.zip
...	...

Listagem com indicação de quantidade de arquivos válidos:

cityname	ctype	n	n_valid
BR-ES-Serra	std	1	0
BR-ES-VilaVelha	std	4	0
...	...	...	...
BR-SP-SaoPaulo	edificacoes	96	96
BR-SP-SaoPaulo	eixos	1	0
BR-SP-SaoPaulo	lotes	96	96
BR-SP-SaoPaulo	planilhas	1	1
...	...	...	...

Listagem com indicação de quantidade de arquivos inválidos:

cityname	ctype	n_nonvalid	errs
BR-ES-Serra	std	1	#ER02: hash not generated;
BR-ES-VilaVelha	std	4	#ER02: hash not generated;
...	...	...	...
BR-SP-SaoPaulo	std	1	#ER01: file extension unknowed; #ER02: hash not generated;

Exemplo de São Paulo

Exemplo de processo de ingestão. As tarefas gerais são:

importar "as is" dos dados geométricos pertinentes (ex. shape files)
importar "as is" dos dados cadastrais pertinentes (ex. planilha IPTU_2020.csv dos endereços)
garantir durante a importação a vinculação com origem (SHA256 do arquivo licenciado).
converter "as is" em dado de importação padronizado, convertendo tipos e nomes de coluna.
consolidar com JOIN para formato padronizado quando for pertinente.

Conforme avaliação dos dados de São Paulo realizada na issue 6 o arquivo IPTU_2020.csv é cadastral (item 2) e, neste exemplo, os dados geométricos (item 1) seriam os shape files ou tabelas denominadas sirgas_shp_lotes_*. O desafio principal é garantir o join (item 5).

Lotes, conforme BR-SP-SaoPaulo-Lotes.shp:

SRID 31983
São identificados pelo ID que é a combinação das strings dos atributos "lo_setor" (length 3), "lo_quadra" (length 3) e "lo_lote" (length 4).
Fórmula para construir a string: lo_setor + lo_quadra + lo_lote

Planilha IPTU 2020:

Única planilha na pasta input/planilhas, IPTU_2020.csv.
Cada entrada é identificada na coluna: "NUMERO DO CONTRIBUINTE" [coluna 0], desprezando o dígito final (exemplo 0050120059-2 vira 0050120059).
O endereço está nas colunas: "NOME DE LOGRADOURO DO IMOVEL" [12], "NUMERO DO IMOVEL" [13]. Opcionais: "BAIRRO DO IMOVEL" [15], "CEP DO IMOVEL" [17]

Eixos de vias, se precisar (SIRGAS_SHP_logradouronbl):

As ruas são identificadas pela combinação das strings dos atributos "lg_tipo", "lg_titulo", "lg_prep" e "lg_nome".
Fórmula para construir a string: lg_tipo + ' ' + lg_titulo + if(lg_titulo, ' ','') + lg_prep + if(lg_prep, ' ','') + lg_nome

Passo 1 - cadastro da origem

CREATE TABLE ingest.origin(
   id serial     NOT NULL PRIMARY KEY,
   fhash text    NOT NULL, -- sha256 is a finger print
   cityname text NOT NULL, -- city name
   fname text    NOT NULL,  -- filename
   fversion smallint NOT NULL DEFAULT 1, -- version counter
   ctype text, -- content type
   is_valid boolean,
   fmeta jsonb,
   ingest_instant timestamp DEFAULT now(),
   UNIQUE(fhash),
   UNIQUE(cityname,fname,fversion) -- ,kx_ingest_date=ingest_instant::date
);
-- EXEMPLO:
INSERT INTO ingest.origin(fhash,cityname, fname, ctype, is_valid, fmeta)
 SELECT fmeta->>'hash', cityname, fname, ctype, is_valid, fmeta - 'hash'
 FROM ingest.cityfolder_input('/home/igor') 
 WHERE is_valid
ON CONFLICT DO NOTHING;

Passo 2 - Carga de dados cadastrais

Por ser mais simples, começamos pelos cadastrais. Poderão ser utilizados também para validar ou conferir a completeza dos dados geométricos.

mkdir -p /tmp/pg_io/br_sp_sp
cd /tmp/pg_io/br_sp_sp
dtrx /home/igor/BR-SP-SaoPaulo/input/planilhas/IPTU_2020.zip # h
recode WINDOWS-1252..UTF-8 IPTU_2020.csv
dos2unix IPTU_2020.csv
# roda gerador de leitor de CSV que cria SQL fwd de leitura
# no psql conferir foreign servers com \des+

-- CREATE EXTENSION IF NOT EXISTS file_fdw;
-- CREATE SERVER    IF NOT EXISTS files FOREIGN DATA WRAPPER file_fdw;
-- CREATE SCHEMA tmp_sp_spa2020; -- namespace for encapsulate ingestion

CREATE FOREIGN TABLE tmp_sp_spa2020.fwd_sp_sp_lotes_csv(
 "NUMERO DO CONTRIBUINTE" text, 
 "ANO DO EXERCICIO" text, 
 "NUMERO DA NL" text, 
 "DATA DO CADASTRAMENTO" text, 
 "TIPO DE CONTRIBUINTE 1" text, 
 "CPF/CNPJ DO CONTRIBUINTE 1" text, 
 "NOME DO CONTRIBUINTE 1" text, 
 "TIPO DE CONTRIBUINTE 2" text, 
 "CPF/CNPJ DO CONTRIBUINTE 2" text, 
 "NOME DO CONTRIBUINTE 2" text, 
 "NUMERO DO CONDOMINIO" text, 
 "CODLOG DO IMOVEL" text, 
 "NOME DE LOGRADOURO DO IMOVEL" text, 
 "NUMERO DO IMOVEL" text, 
 "COMPLEMENTO DO IMOVEL" text, 
 "BAIRRO DO IMOVEL" text, 
 "REFERENCIA DO IMOVEL" text, 
 "CEP DO IMOVEL" text, 
 "QUANTIDADE DE ESQUINAS/FRENTES" text, 
 "FRACAO IDEAL" text, 
 "AREA DO TERRENO" text, 
 "AREA CONSTRUIDA" text, 
 "AREA OCUPADA" text, 
 "VALOR DO M2 DO TERRENO" text, 
 "VALOR DO M2 DE CONSTRUCAO" text, 
 "ANO DA CONSTRUCAO CORRIGIDO" text, 
 "QUANTIDADE DE PAVIMENTOS" text, 
 "TESTADA PARA CALCULO" text, 
 "TIPO DE USO DO IMOVEL" text, 
 "TIPO DE PADRAO DA CONSTRUCAO" text, 
 "TIPO DE TERRENO" text, 
 "FATOR DE OBSOLESCENCIA" text, 
 "ANO DE INICIO DA VIDA DO CONTRIBUINTE" text, 
 "MES DE INICIO DA VIDA DO CONTRIBUINTE" text, 
 "FASE DO CONTRIBUINTE" text
) SERVER files OPTIONS (
        filename '/tmp/pg_io/br_sp_sp/IPTU_2020.csv',
        format 'csv',
        header 'true',
        delimiter ';'
);
-- teste de volumetria:
select count(*) from tmp_sp_spa2020.fwd_sp_sp_lotes_csv; -- 3498644
-- teste de composição dos dados
SELECT "NUMERO DO IMOVEL" obj_id, 
       regexp_replace("NUMERO DO CONTRIBUINTE", '\-\d+$', '') obj_owner_id,
       "NOME DE LOGRADOURO DO IMOVEL" as obj_streetname,
       "NUMERO DO IMOVEL" as obj_housenum,
       "CEP DO IMOVEL" as cep
       -- , "BAIRRO DO IMOVEL" --col [15]
FROM tmp_sp_spa2020.fwd_sp_sp_lotes_csv
LIMIT 100;

Amostrando algumas das 3498644 linhas desta tabela:

obj_id	obj_owner_id	obj_streetname	obj_housenum	cep
13	0010030001	R S CAETANO	13	01104-001
19	0010030002	R S CAETANO	19	01104-001
27	0010030003	R S CAETANO	27	01104-001
33	0010030004	R S CAETANO	33	01104-001
...	...	...	...	...

Passo 3 - Carga de dados geométricos

A carga por shape file requer conhecimento a priori do SRID (SIRGAS é 31983) e dos campos a serem utilizados.

SELECT gid, lo_setor || lo_quadra || lo_lote as obj_id, st_area(geom) as area, st_area(st_transform(geom,4326),true) as area2
FROM sirgas_shp_lotes_01_agua_rasa;

(wiki temporária!)

Navigation Menu

Uh oh!

Upload e integridade dos dados originais

Carga para o servidor

Avaliação dos arquivos carregados:

Convenções de nomes de pasta e subpasta e tipos

Join com sha258 da origem

Exemplo de São Paulo

Passo 1 - cadastro da origem

Passo 2 - Carga de dados cadastrais

Passo 3 - Carga de dados geométricos

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally