+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:
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')
0 Comments