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	Descrição do campo
MÊS	MES	DOUBLE	Tipo de Documento de Vendas
N° PESSOAL	PERNR	BIGINT	Organização de Vendas
NOME COMPLETO	CNAME	STRING	Equipe de Vendas
EMPRESA	BUKRS_D	STRING	Escritório de Vendas
NOME DA FIRMA	WERKS	STRING	Região de Vendas
ÁREA RECURSOS HUMANOS	WERKS_D	STRING	Condição de Pagamento
DESCRIÇÃO COMPLETA ÁREA RH	BTRTL	STRING	Condição de Expedição
SUBÁREA REC.HUMANOS	BTRTL_D	STRING	Moeda do documento SD
TXT.SUBÁREA REC.HUM.	GSBER	STRING	Entrega Paletizada
DIVISÃO	GSBER_D	STRING	Incoterms parte 1
DENOMINAÇÃO DA DIVISÃO	PERSG	STRING	Incoterms parte 2
GRUPO EMPREGADOS	PERSG_D	STRING	Setor de Atividade
DENOM.GRP.EMPREG.	PERSK	STRING	Canal de Distribuição
SUBGRUPO EMPREGADOS	PERSK_D	STRING	Louça Embalada
PARTN_ROLE	ABKRS	STRING	Função do Parceiro
	ABKRS_D	STRING	Utilização
	ORGEH	BIGINT	Emissor da Ordem
	ORGEH_A_SUP	STRING	Centro
CUST_MAT	ORGEH_D	STRING	Material do Cliente
	STELL	BIGINT
	STELL_D	STRING
	KOSTL	STRING
	KOSTL_D	STRING
	FAMST	DOUBLE
	FAMST_D	STRING
	GESCH	BIGINT
	GESCH_D	STRING
	GBDAT	TIMESTAMP
	GBDAT_IDADE	BIGINT
	CHEFE_PERNR	BIGINT
	CHEFE_CNAME	STRING
	GERENTE_PERNR	BIGINT
	GERENTE_CNAME	STRING
	DIRETOR_PERNR	BIGINT
	DIRETOR_CNAME	STRING
	HRS_PREV	DOUBLE
	FALTAS_INJUST	DOUBLE
	FALTA_ABON	DOUBLE
	FALTAS_JUST	DOUBLE
	FALTAS_LEGAIS	DOUBLE
	ATESTADOS	DOUBLE
	AFASTAMENTOS	DOUBLE
	FALTAS_SEM_AFAST	DOUBLE
	TOT_FALTAS	DOUBLE
	PERC_FALTAS_AFAST	DOUBLE
	PERC_FALTAS_SEM_AFAST	DOUBLE
	PERC_ABSENT_GERAL	DOUBLE
	ARTIGO59	BIGINT
	ARTIGO66	BIGINT
	ARTIGO67	BIGINT
	ARTIGO77	BIGINT
	HE_A_DEFINIR	DOUBLE
	FALTAS_A_DEFINIR	DOUBLE
	TOT_HE_REALIZADA	DOUBLE
	SLD_ANT_BH	DOUBLE
	SLD_ANT_NEG	DOUBLE
	SLD_ANT_NEG_EMPR	DOUBLE
	SLD_ANT_NEG_COVID19	DOUBLE
	SLD_ANT_EMPREGADO	DOUBLE
	BH_CREDITO	DOUBLE
	BH_DEBITO	DOUBLE
	BH_DEB_COVID19	DOUBLE
	BH_DEB_EMPREGADO	DOUBLE
	BH_MES	DOUBLE
	BH_ACUM	DOUBLE
	BH_ACUM_NEG	DOUBLE
	BH_NEG_EMPR	DOUBLE
	BH_NEG_COVID19	DOUBLE
	BH_ANT_EMPREGADO	DOUBLE
	PAGTO_BH	DOUBLE
	DESC_BH	DOUBLE
	PERDAO_SLD_NEG	DOUBLE
	HE_FERIADOS	DOUBLE
	HE_TRANSP	DOUBLE
	HE_INCEND	DOUBLE
	HE_FERIADOS_TURNO	DOUBLE
	HE_FOLGAS	DOUBLE
	HE_DIA_NORMAL	DOUBLE
	HE_PAGAS	DOUBLE
	.OCORRENCIAS	BIGINT
	SALDO_PERDAO_BH_COVID	DOUBLE
	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')