Recursos humanos - Carga de arquivos na zitrhr145

+Documentação do código

Com o SAP o ultilizamos para pegar os dados envia-los para o exel, para conseguir verificar todos os dados separados por linhas e colunas. Utilizando o excel nós observamos e verificamos todos os arquivos para ver se todas as colunas batiam, se não tinha uma faltando ou sobrando. Blob Storage serveiu como um armazenamento de dados, dentro do proprio databricks, pegamos os arquivos e colocamos dentro de uma pasta no databricks para utiliza-los no código. Pyspark foi usado para fazer toda a codificação do arquivo, pegando tabela por tabela e as juntando para se tranformarem em uma só, definindo tipos para as colunas e classificações, juntamente com o databricks que é a plataforma onde foi feita essa codificação. Data Bricks usado para grande movimentação e edição de dados.

Desenvolvimento da carga de arquivos

Instalação de bibliotecas

A instalação desta biblioteca, serve para trazer a possibilidade de ler dados, e informações do Excel. O Python no código fonte não tem essa função então a instalamos. Conseguimos assim puxar arquivos xlsx para dentro do databricks e ver os seus dados. A instalação desta biblioteca, serve para nós conseguirmos ler e gravar arquivos no excel.

pip install xlrd
pip install openpyxl

Importação de bibliotecas

“Import os”: serve para importar a interface do sistema operacional para funções que dependem dele. “import pandas as pd”: importando a biblioteca pandas para fazermos as análises de dados. “from”: de onde vamos importar para o código os tipos e funções.

import os
import pandas as pd
from pyspark.sql.types import DoubleType, StringType, LongType, TimestampType, StructType, StructField
from pyspark.sql.functions import col

Leitura e agregação

Fazendo a chamada dos arquivos xlsx para usarmos no código, “.listdir” é usado para obter a lista de todos os arquivos e diretórios no diretório especificado. “.DataFrame” usado para fazer a organização dos dados para a visualização. Após isso é usado o “for” para pegar todos os dados de um arquivo xlsx e adicionar a outro arquivo assim ficam juntos em apenas um só e esse “for” é repetido o número a quantidade que tem de arquivos xlsx, nesse caso se repete 24 vezes esse processo até sobrar apenas 1 arquivo com todos os dados dos 23 outros. “endswith” serve para pegar todos os dados que terminem com o determinado sulfixo. “Append” usado para adicionar algum dado dentro de algum lugar, nesse caso, o utilizamos para adicionar os dados de um arquivo dentro de outro, puxamos a tabela x pegamos os dados da tabela y e esses dados irão para a tabela y e assim por diante.

cwd = os.path.abspath('/dbfs/FileStore/recursos_humanos_gerencial_time_zitrhr145') 
files = os.listdir(cwd)  
df_python = pd.DataFrame()
for file in files:
    if file.endswith('.xlsx'):
        df_python= df_python.append(pd.read_excel(cwd+'/'+file), ignore_index=True)

Criação do schema

Schema é a estrutura do data frame, estamos pegando coluna por coluna e dando a ela um tipo e quando quisermos verificar o código ele estará estruturado com os dados certos. “stringtype”, “longtype” serve para tipar o dado, ou seja, dizer se ele é um número normal ou grande, uma string.

schema = StructType([ \
    StructField('MES', StringType(),True),
    StructField('PERNR', LongType(),True),
    StructField('CNAME', StringType(),True),
    StructField('BUKRS', StringType(),True),
    StructField('BUKRS_D', StringType(),True),
    StructField('WERKS', StringType(),True),
    StructField('WERKS_D', StringType(),True),
    StructField('BTRTL', StringType(),True),
    StructField('BTRTL_D', StringType(),True),
    StructField('GSBER', StringType(),True),
    StructField('GSBER_D', StringType(),True),
    StructField('PERSG', StringType(),True),
    StructField('PERSG_D', StringType(),True),
    StructField('PERSK', StringType(),True),
    StructField('PERSK_D', StringType(),True),
    StructField('ABKRS', StringType(),True),
    StructField('ABKRS_D', StringType(),True),
    StructField('ORGEH', LongType(),True),
    StructField('ORGEH_A_SUP', StringType(),True),
    StructField('ORGEH_D', StringType(),True),
    StructField('STELL', LongType(),True),
    StructField('STELL_D', StringType(),True),
    StructField('KOSTL', StringType(),True),
    StructField('KOSTL_D', StringType(),True),
    StructField('FAMST', StringType(),True),
    StructField('FAMST_D', StringType(),True),
    StructField('GESCH', LongType(),True),
    StructField('GESCH_D', StringType(),True),
    StructField('GBDAT', StringType(),True),
    StructField('GBDAT_IDADE', LongType(),True),
    StructField('CHEFE_PERNR', LongType(),True),
    StructField('CHEFE_CNAME', StringType(),True),
    StructField('GERENTE_PERNR', LongType(),True),
    StructField('GERENTE_CNAME', StringType(),True),
    StructField('DIRETOR_PERNR', LongType(),True),
    StructField('DIRETOR_CNAME', StringType(),True),
    StructField('HRS_PREV', StringType(),True),
    StructField('FALTAS_INJUST', StringType(),True),
    StructField('FALTA_ABON', StringType(),True),
    StructField('FALTAS_JUST', StringType(),True),
    StructField('FALTAS_LEGAIS', StringType(),True),
    StructField('ATESTADOS', StringType(),True),
    StructField('AFASTAMENTOS', StringType(),True),
    StructField('FALTAS_SEM_AFAST', StringType(),True),
    StructField('TOT_FALTAS', StringType(),True),
    StructField('PERC_FALTAS_AFAST', StringType(),True),
    StructField('PERC_FALTAS_SEM_AFAST', StringType(),True),
    StructField('PERC_ABSENT_GERAL', StringType(),True),
    StructField('ARTIGO59', LongType(),True),
    StructField('ARTIGO66', LongType(),True),
    StructField('ARTIGO67', LongType(),True),
    StructField('ARTIGO77', LongType(),True),
    StructField('HE_A_DEFINIR', StringType(),True),
    StructField('FALTAS_A_DEFINIR', StringType(),True),
    StructField('TOT_HE_REALIZADA', StringType(),True),
    StructField('SLD_ANT_BH', StringType(),True),
    StructField('SLD_ANT_NEG', StringType(),True),
    StructField('SLD_ANT_NEG_EMPR', StringType(),True),
    StructField('SLD_ANT_NEG_COVID19', StringType(),True),
    StructField('SLD_ANT_EMPREGADO', StringType(),True),
    StructField('BH_CREDITO', StringType(),True),
    StructField('BH_DEBITO', StringType(),True),
    StructField('BH_DEB_COVID19', StringType(),True),
    StructField('BH_DEB_EMPREGADO', StringType(),True),
    StructField('BH_MES', StringType(),True),
    StructField('BH_ACUM', StringType(),True),
    StructField('BH_ACUM_NEG', StringType(),True),
    StructField('BH_NEG_EMPR', StringType(),True),
    StructField('BH_NEG_COVID19', StringType(),True),
    StructField('BH_ANT_EMPREGADO', StringType(),True),
    StructField('PAGTO_BH', StringType(),True),
    StructField('DESC_BH', StringType(),True),
    StructField('PERDAO_SLD_NEG', StringType(),True),
    StructField('HE_FERIADOS', StringType(),True),
    StructField('HE_TRANSP', StringType(),True),
    StructField('HE_INCEND', StringType(),True),
    StructField('HE_FERIADOS_TURNO', StringType(),True),
    StructField('HE_FOLGAS', StringType(),True),
    StructField('HE_DIA_NORMAL', StringType(),True),
    StructField('HE_PAGAS', StringType(),True),
    StructField('HRS_TRAB', StringType(),True),
    StructField('OCORRENCIAS', LongType(),True),
    StructField('SALDO_PERDAO_BH_COVID', StringType(),True),
    StructField('SALDO_PERDAO_BH_EMPRESA', StringType(), True)])

Criação do data frame

“Spark.conf.set” serve para definir vários parâmetros do Spark como pares de valores-chave. “CreateDataFrame” está criando o data frame para a verificação dos dados conforme a estrutura do schema e os dados do df_python.

 spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "false")
sc = spark.createDataFrame(df_python, schema)

Alterção de dado

Estamos selecionando todas as colunas com o “col”. O “.cast” ele serve para converter o tipo de dado para outro, nesse caso estamos convertendo para representar valores duplos, um inteiro grande com grande quantidade de caracteres.

df_final = sc.select(col('MES').cast('double'),
                     col('PERNR').cast('bigint'),
                     col('CNAME').cast('string'),
                     col('BUKRS').cast('string'),
                     col('BUKRS_D').cast('string'),
                     col('WERKS').cast('string'),
                     col('WERKS_D').cast('string'),
                     col('BTRTL').cast('string'),
                     col('BTRTL_D').cast('string'),
                     col('GSBER').cast('string'),
                     col('GSBER_D').cast('string'),
                     col('PERSG').cast('string'),
                     col('PERSG_D').cast('string'),
                     col('PERSK').cast('string'),
                     col('PERSK_D').cast('string'),
                     col('ABKRS').cast('string'),
                     col('ABKRS_D').cast('string'),
                     col('ORGEH').cast('bigint'),
                     col('ORGEH_A_SUP').cast('string'),
                     col('ORGEH_D').cast('string'),
                     col('STELL').cast('bigint'),
                     col('STELL_D').cast('string'),
                     col('KOSTL').cast('string'),
                     col('KOSTL_D').cast('string'),
                     col('FAMST').cast('double'),
                     col('FAMST_D').cast('string'),
                     col('GESCH').cast('bigint'),
                     col('GESCH_D').cast('string'),
                     col('GBDAT').cast('timestamp'),
                     col('GBDAT_IDADE').cast('bigint'),
                     col('CHEFE_PERNR').cast('bigint'),
                     col('CHEFE_CNAME').cast('string'),
                     col('GERENTE_PERNR').cast('bigint'),
                     col('GERENTE_CNAME').cast('string'),
                     col('DIRETOR_PERNR').cast('bigint'),
                     col('DIRETOR_CNAME').cast('string'),
                     col('HRS_PREV').cast('double'),
                     col('FALTAS_INJUST').cast('double'),
                     col('FALTA_ABON').cast('double'),
                     col('FALTAS_JUST').cast('double'),
                     col('FALTAS_LEGAIS').cast('double'),
                     col('ATESTADOS').cast('double'),
                     col('AFASTAMENTOS').cast('double'),
                     col('FALTAS_SEM_AFAST').cast('double'),
                     col('TOT_FALTAS').cast('double'),
                     col('PERC_FALTAS_AFAST').cast('double'),
                     col('PERC_FALTAS_SEM_AFAST').cast('double'),
                     col('PERC_ABSENT_GERAL').cast('double'),
                     col('ARTIGO59').cast('bigint'),
                     col('ARTIGO66').cast('bigint'),
                     col('ARTIGO67').cast('bigint'),
                     col('ARTIGO77').cast('bigint'),
                     col('HE_A_DEFINIR').cast('double'),
                     col('FALTAS_A_DEFINIR').cast('double'),
                     col('TOT_HE_REALIZADA').cast('double'),
                     col('SLD_ANT_BH').cast('double'),
                     col('SLD_ANT_NEG').cast('double'),
                     col('SLD_ANT_NEG_EMPR').cast('double'),
                     col('SLD_ANT_NEG_COVID19').cast('double'),
                     col('SLD_ANT_EMPREGADO').cast('double'),
                     col('BH_CREDITO').cast('double'),
                     col('BH_DEBITO').cast('double'),
                     col('BH_DEB_COVID19').cast('double'),
                     col('BH_DEB_EMPREGADO').cast('double'),
                     col('BH_MES').cast('double'),
                     col('BH_ACUM').cast('double'),
                     col('BH_ACUM_NEG').cast('double'),
                     col('BH_NEG_EMPR').cast('double'),
                     col('BH_NEG_COVID19').cast('double'),
                     col('BH_ANT_EMPREGADO').cast('double'),
                     col('PAGTO_BH').cast('double'),
                     col('DESC_BH').cast('double'),
                     col('PERDAO_SLD_NEG').cast('double'),
                     col('HE_FERIADOS').cast('double'),
                     col('HE_TRANSP').cast('double'),
                     col('HE_INCEND').cast('double'),
                     col('HE_FERIADOS_TURNO').cast('double'),
                     col('HE_FOLGAS').cast('double'),
                     col('HE_DIA_NORMAL').cast('double'),
                     col('HE_PAGAS').cast('double'),
                     col('HRS_TRAB').cast('double'),
                     col('OCORRENCIAS').cast('bigint'),
                     col('SALDO_PERDAO_BH_COVID').cast('double'),
                     col('SALDO_PERDAO_BH_EMPRESA').cast('double')
)

Demonstração dos arquivos

Os arquivos extraidos para o blob storage foram estes:

Descrição das colunas

A descrição das colunas da tabela zitrhr145 é conforme abaixo:

Campos Origem SAP	Campos Extraídos	Data Types
MÊS	MES	DOUBLE
N° PESSOAL	PERNR	BIGINT
NOME COMPLETO	CNAME	STRING
EMPRESA	BUKRS	STRING
NOME DA FIRMA	BUKRS_D	STRING
ÁREA RECURSOS HUMANOS	WERKS	STRING
DESCRIÇÃO COMPLETA ÁREA RH	WERKS_D	STRING
SUBÁREA REC.HUMANOS	BTRTL	STRING
TXT.SUBÁREA REC.HUM.	BTRTL_D	STRING
DIVISÃO	GSBER	STRING
DENOMINAÇÃO DA DIVISÃO	GSBER_D	STRING
GRUPO EMPREGADOS	PERSG	STRING
DENOM.GRP.EMPREG.	PERSG_D	STRING
SUBGRUPO EMPREGADOS	PERSK	STRING
DENOM.SUBGRP.EMPRG	PERSK_D	STRING
ÁREA PROC.FLHPAGTO	ABKRS	STRING
TXT.ÁREA PROCESS.FP	ABKRS_D	STRING
UNID.ORGANIZACIONAL	ORGEH	BIGINT
ABREVIAÇÃO UNID. ORG. SUPERIOR	ORGEH_A_SUP	STRING
UNIDADEORGANIZ.	ORGEH_D	STRING
CHAVE DO CARGO	STELL	BIGINT
DENOMINAÇÃO DE CARGO	STELL_D	STRING
CENTRO DE CUSTO	KOSTL	STRING
DESCRIÇÃO CENTRO CUSTO	KOSTL_D	STRING
ESTADO CIVIL	FAMST	DOUBLE
ESTADO CIVIL	FAMST_D	STRING
SEXO	GESCH	BIGINT
DENOMINAÇÃO SEXO	GESCH_D	STRING
DATA DE NASCIMENTO	GBDAT	TIMESTAMP
IDADE	GBDAT_IDADE	BIGINT
MATRÍCULA CHEFE	CHEFE_PERNR	BIGINT
NOME CHEFE	CHEFE_CNAME	STRING
MATRÍCULA GERENTE	GERENTE_PERNR	BIGINT
NOME GERENTE	GERENTE_CNAME	STRING
MATRÍCULA DIRETOR	DIRETOR_PERNR	BIGINT
NOME DIRETOR	DIRETOR_CNAME	STRING
HORAS PREVISTAS	HRS_PREV	DOUBLE
FALTAS INJUSTIFICADAS	FALTAS_INJUST	DOUBLE
FALTAS ABONADAS	FALTA_ABON	DOUBLE
FALTAS JUSTIFICADAS	FALTAS_JUST	DOUBLE
FALTAS LEGAIS	FALTAS_LEGAIS	DOUBLE
ATESTADOS	ATESTADOS	DOUBLE
AFASTAMENTOS	AFASTAMENTOS	DOUBLE
FALTAS SEM AFASTAMENTOS	FALTAS_SEM_AFAST	DOUBLE
TOTAL DE FALTAS	TOT_FALTAS	DOUBLE
% FALTAS AFASTAMENTOS	PERC_FALTAS_AFAST	DOUBLE
% FALTAS SEM AFASTAMENTOS	PERC_FALTAS_SEM_AFAST	DOUBLE
% ABSENTEÍSMO GERAL	PERC_ABSENT_GERAL	DOUBLE
ARTIGO 59	ARTIGO59	BIGINT
ARTIGO 66	ARTIGO66	BIGINT
ARTIGO 67	ARTIGO67	BIGINT
ARTIGO 77	ARTIGO77	BIGINT
HE A DEFINIR	HE_A_DEFINIR	DOUBLE
FALTAS A DEFINIR	FALTAS_A_DEFINIR	DOUBLE
TOTAL HE REALIZADA	TOT_HE_REALIZADA	DOUBLE
SALDO ANTERIOR BH	SLD_ANT_BH	DOUBLE
SALDO ANTERIOR NEGATIVO	SLD_ANT_NEG	DOUBLE
SALDO ANT.NEG.EMPR	SLD_ANT_NEG_EMPR	DOUBLE
SALDO ANT.NEG.COVID-19	SLD_ANT_NEG_COVID19	DOUBLE
SALDO ANT. EMPREGADO	SLD_ANT_EMPREGADO	DOUBLE
BH CREDITO	BH_CREDITO	DOUBLE
BH DÉBITO EMPRESA	BH_DEBITO	DOUBLE
BH DÉBITO COVID-19	BH_DEB_COVID19	DOUBLE
BH DÉBITO EMPREGADO	BH_DEB_EMPREGADO	DOUBLE
BH MÊS	BH_MES	DOUBLE
BH ACUMULADO	BH_ACUM	DOUBLE
BH ACUMULADO NEGATIVO	BH_ACUM_NEG	DOUBLE
BH NEGATIVO EMPRESA	BH_NEG_EMPR	DOUBLE
BH NEGATIVO COVID-19	BH_NEG_COVID19	DOUBLE
BH NEGATIVO EMPREGADO	BH_ANT_EMPREGADO	DOUBLE
PAGAMENTO BH	PAGTO_BH	DOUBLE
DESCONTO BH	DESC_BH	DOUBLE
PERDÃO DE SALDO NEGATIVO	PERDAO_SLD_NEG	DOUBLE
HE PAGA FERIADO	HE_FERIADOS	DOUBLE
HORA EXTRA TRANSPORTE	HE_TRANSP	DOUBLE
HORA EXTRA INCÊNDIO	HE_INCEND	DOUBLE
HE PAGA FERIADO TURNO	HE_FERIADOS_TURNO	DOUBLE
HE PAGA FOLGA	HE_FOLGAS	DOUBLE
HE PAGA DIA NORMAL	HE_DIA_NORMAL	DOUBLE
TOTAL HE PAGA	HE_PAGAS	DOUBLE
HORAS TRABALHADAS	.OCORRENCIAS	BIGINT
SALDO PERDÃO BH COVID	SALDO_PERDAO_BH_COVID	DOUBLE
SALDO PERDÃO BH EMPRESA	SALDO_PERDAO_BH_EMPRESA	DOUBLE

Gravação do código

Após realizar todos os comandos necessário é dado um “.write” para finalizar a codificação, para salva-la. Dizendo o modo que vai ser efetuada esse salvamento se vai sobreescrever o anterior, apagar, acrescentar dados, nesse caso usado o “append” pois iremos juntar todos os dados. Salvar na tabela que queremos a informação.

df_final.write.format('delta').mode('append').saveAsTable('recursos_humanos.zitrhr145_copy_teste')