/
Dados Internos Dexco

Conjunto de notebooks que tem como objetivo fazer a ingestão de dados e indicadores de diversas áreas da Dexco e disponibilizar da forma mais automatizada possível para serem consumidos pelo dashboard desenvolvido para o projeto.

Índice

  1. Indicadores Carteira Deca

  2. Indicadores Devolução Deca

  3. Indicadores Devolução Madeira

  4. Indicadores Financeiros

  5. Indicadores Gente

  7. Indicadores Margem Madeira

  9. Indicadores OEE Madeira

  10. Indicadores OTIF Deca

  11. Indicadores OTIF Madeira

  12. Indicadores Panorama Mercado Madeira

  13. Indicadores Produtividade Deca

  14. Indicadores SAC

  15. Indicadores Sell-in Deca - Chuveiros

  16. Indicadores Sell-in Deca - Cubas

  17. Indicadores Sell-in Deca - Torneiras

  18. Indicadores Market Share, Sell-in e Sell-out Madeira

  19. Indicadores Vendas Deca

1. Indicadores Carteira Deca

Indicadores para pedidos em carteira Deca.

  • Notebook: indicadores_carteira_deca

  • Job: inteligencia-mercado_job_prod_indicadores_carteira_deca

  • Schedule: diário/18:00h

  • Base fim: indicadores_mercado.tb_indicadores_carteira_deca

1.1 Origem:

Database DEV do Redshift (dtx-deca-sellin.czcbob9woqfg.us-east-1.redshift.amazonaws.com), schema small e tabela tb_ordem_pendente.

1.2 Transformação:

Fazendo uma query com um select de todos os campos da tabela origem:

1query = "select * from small.tb_ordem_pendente" 2multiple_run_parameters = dbutils.notebook.entry_point.getCurrentBindings() 3bucket_name = multiple_run_parameters["bucket_name"] 4pedidos_carteira = redshift_to_dataframe(query = query, filename = "tb_ordem_pendente", bucket_name = bucket_name)

Renomeando as colunas:

1colunas_pedidos_carteira = ["codigo_empresa", "numero_ordem_venda", "numero_sequencia_item_ordem_venda", "codigo_organizacao_vendas", "codigo_canal_distribuicao", "codigo_setor_atividade", "codigo_escritorio_vendas", "codigo_equipe_vendas", "codigo_emissor_ordem", "data_primeira_remessa", "data_emissao", "data_pedido_cliente", "motivo_recusa", "tipo_documento_ordem_venda", "quantidade_itens", "quantidade_faturada_ordem", "quantidade_pendente", "valor_liquido", "valor_faturado_ordem", "valor_pendente", "status_faturamento", "bloqueio_remessa_cliente", "status_verificacao_credito", "status_carteira", "codigo_produto", "descricao_produto", "data_atualizacao", "remessa", "data_desejada_remessa", "status_recusa", "status_item", "codigo_centro"] 2df_pedidos_carteira = pedidos_carteira.toDF(*colunas_pedidos_carteira)

Utilizando Spark SQL para fazer alguns ajustes e aplicar algumas regras de negócio:

1df_carteira = spark.sql(""" 2 select 3 to_date(data_atualizacao) as data_referencia, 4 (case when codigo_setor_atividade == 'HY' then 'hydra' 5 when codigo_setor_atividade == 'MS' then 'metais' 6 when codigo_setor_atividade == 'CS' then 'loucas' end) as negocio, 7 sum(case when status_carteira in('Bloqueio Adm.', 'Credito', 'Limbo', 'Limbo Programado', 'Não classificado') then valor_pendente else 0 end) as valor_bloqueado, 8 sum(case when status_carteira in('Programado', 'Remetido') then valor_pendente else 0 end) as valor_livre 9 10 from pedidos_carteira 11 12 where to_date(data_atualizacao) == current_date() 13 and codigo_setor_atividade in('CS', 'HY', 'MS') 14 15 group by 1, 2 16""")

Fazendo append na tabela final na database indicadores_mercado:

1df_carteira.write.mode("append").saveAsTable("indicadores_mercado.tb_indicadores_carteira_deca")

1.3 Base Final:

col_name

data_type

data_referencia

date

negocio

string

valor_bloqueado

double

valor_livre

double


2. Indicadores Devolução Deca

Indicadores para pedidos em status de devolução para Deca.

  • Notebook: indicadores_devolucao_deca

  • Job: inteligencia-mercado_job_prod_indicadores_devolucao_deca

  • Schedule: diário/14:00h

  • Base fim: indicadores_mercado.tb_indicadores_devolucoes_deca

2.1 Origem:

Database large e tabela tb_resultado_comercial.

2.2 Transformação:

Utilizando o Spark SQL para query da tabela origem já aplicando regras de negócio:

1df_devolucoes_deca = spark.sql(""" 2 select distinct 3 to_date(date_format(data_competencia, 'yyyy-MM-01')) as mes, 4 (case when codigo_setor_atividade == 'MS' then 'metais' 5 when codigo_setor_atividade == 'CS' then 'loucas' 6 when codigo_setor_atividade == 'HY' then 'hydra' 7 when codigo_setor_atividade == '01' then 'revestimento' end) as negocio, 8 sum(case when tipo_documento_venda == 'S2' then valor_receita_liquida else 0 end) as valor_estorno, 9 sum(case when status_ordem_venda == 'DEVOLUÇÃO' then abs(valor_receita_liquida) else 0 end) as valor_devolucao, 10 sum(case when status_ordem_venda == 'FATURAMENTO' then valor_receita_liquida else 0 end) as valor_faturamento 11 12 from large.tb_resultado_comercial 13 14 where data_competencia between '2019-01-01' and current_date() - 1 15 and codigo_setor_atividade in('MS', 'CS', 'HY', '01') 16 17 group by 1, 2 18 19 order by 1, 2 asc 20""")

Calculando o valor de devolução ajustado, conforme regra de negócio:

1df_devolucoes_deca = df_devolucoes_deca.withColumn("valor_devolucao_ajustado", col("valor_devolucao") - col("valor_estorno"))

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_devolucoes_deca.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_devolucoes_deca")

2.3 Base Final:

col_name

data_type

mes

date

negocio

string

valor_faturamento

double

valor_devolucao

double


3. Indicadores Devolução Madeira

Indicadores para pedidos em status de devolução para Madeira.

  • Notebook: indicadores_devolucao_madeira

  • Job: inteligencia-mercado_job_prod_indicadores_devolucao_madeira

  • Schedule: diário/12:00h

  • Base fim: indicadores_mercado.tb_indicadores_devolucoes_madeira

3.1 Origem:

Database analytics_prd e tabela custos_rem.

3.2 Transformação:

Utilizando o Spark SQL para query da tabela origem já aplicando regras de negócio:

1df_devolucao_madeira = spark.sql(""" 2 select 3 perio as mes, 4 (case when prctr like '%MDP' then 'mdp' 5 when prctr like '%MDF' then 'mdf' else 'paineis' end) as negocio, 6 sum(case when fkart in('ZREB', 'ZROB') then abs((vv089) - (vv001 + vv002 + vv003 + vv004)) else 0 end) as valor_devolucao, 7 sum(case when fkart not in('ZREB', 'ZROB') then (vv089) - (vv001 + vv002 + vv003 + vv004) else 0 end) as valor_faturado 8 9 from analytics_prd.custos_rem 10 11 where spart == 'CH' 12 and (prctr like '%MDP' or prctr like '%MDF') 13 14 and perio >= 2019001 15 16 group by 1, 2 17 18 order by 1, 2 19""")

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_devolucao_madeira.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_devolucoes_madeira")

3.3 Base Final:

col_name

data_type

mes

date

negocio

string

valor_faturamento

decimal(29,2)

valor_devolucao

decimal(29,2)


4. Indicadores Financeiros

Indicadores financeiros disponibilizados pela Controladoria.

  • Notebook: indicadores_financeiros

  • Job: inteligencia-mercado_job_prod_indicadores_financeiros

  • Schedule: diário/15:00h

  • Base fim: indicadores_mercado.tb_indicadores_financeiros_2022

4.1 Origem:

A Controladoria atualiza mensalmente alguns arquivos Excel disponibilizados em no Sharepoint Relatório Controladoria.

4.2 Transformação:

Foi desenvolvida uma função para fazer o download desses arquivos e gravar no diretório do dbfs:

1def download_arquivo(arquivo_download, nome_arquivo): 2 ctx = ClientContext(site_sharepoint).with_credentials(UserCredential(usuario, senha)) 3 web = ctx.load(ctx.web).execute_query() 4 response = File.open_binary(ctx, arquivo_download) 5 response.raise_for_status() 6 with open("/dbfs/FileStore/shared_uploads/arquivos_financeiros/" + nome_arquivo, "wb") as pasta: 7 pasta.write(response.content)

Fazendo o download dos arquivos:

1download_arquivo("/sites/RelatriosControladoria/Documentos%20Compartilhados/Report%20Book%20Diretoria/Fechamento%20Gerencial%202022.xlsm", "Fechamento_Gerencial_2022.xlsm") 2download_arquivo("/sites/RelatriosControladoria/Documentos%20Compartilhados/Report%20Book%20Diretoria/Fluxo%20de%20Caixa%20Oficial%202022.xlsx", "Fluxo_de_Caixa_Oficial_2022.xlsx") 3download_arquivo("/sites/RelatriosControladoria/Documentos%20Compartilhados/Report%20Book%20Diretoria/Forecast%20Consolidador.xlsx", "Forecast_3_9_2022.xlsx") 4download_arquivo("/sites/RelatriosControladoria/Documentos%20Compartilhados/Report%20Book%20Diretoria/Forecast%20-%20Fluxo%20de%20Caixa%20Livre.xlsx", "FCST_2_10___Fluxo_de_Caixa_Livre_2022.xlsx")

4.2.1 Arquivo Fechamento_Gerencial_2022.xlsm

4.2.1.1 Indicadores do DRE

Definindo função para Indicadores do DRE:

1#parâmetros da função: range de células e negócio("madeira", "deca", "revestimento" ou "consolidado") 2def ler_dre(celulas, negocio): 3 #lendo range de células do excel utilizando o pandas.read_excel(arquivo, sheet, linhas desconsideradas, número de linhas, range células, coluna index) 4 #transpondo a tabela utilizando .transpose() 5 #selecionando colunas utilizando .iloc[] 6 #resetando index 7 df = pd.read_excel(fechamento_gerencial, "DRE CONSOL", skiprows = 1, nrows = 146, usecols = celulas, index_col = 0).transpose().iloc[:, [6, 127]].reset_index() 8 #renomeando colunas 9 df.columns = dre_header 10 #transformando em dataframe do spark 11 df = spark.createDataFrame(df) 12 #criando coluna "negocio" 13 df = df.selectExpr(dre_header[0], "'{}' as negocio".format(negocio), dre_header[1], dre_header[2]) 14 #tratando coluna "mes" 15 df = df.withColumn("mes", when(df.mes.like("JAN%"), "01/01/" + ano_atual) 16 .when(df.mes.like("FEV%"), "01/02/" + ano_atual) 17 .when(df.mes.like("MAR%"), "01/03/" + ano_atual) 18 .when(df.mes.like("ABR%"), "01/04/" + ano_atual) 19 .when(df.mes.like("MAI%"), "01/05/" + ano_atual) 20 .when(df.mes.like("JUN%"), "01/06/" + ano_atual) 21 .when(df.mes.like("JUL%"), "01/07/" + ano_atual) 22 .when(df.mes.like("AGO%"), "01/08/" + ano_atual) 23 .when(df.mes.like("SET%"), "01/09/" + ano_atual) 24 .when(df.mes.like("OUT%"), "01/10/" + ano_atual) 25 .when(df.mes.like("NOV%"), "01/11/" + ano_atual) 26 .when(df.mes.like("DEZ%"), "01/12/" + ano_atual) 27 .otherwise(None)) 28 df = df.withColumn("mes", to_date("mes", "dd/MM/yyyy")) 29 return df

Aplicando a função para cada negócio, passando o parâmetro de range de células onde se encontram os dados:

1df_dre_madeira = ler_dre("EL:EX", "madeira") 2df_dre_deca = ler_dre("B:N", "deca") 3df_dre_revestimento = ler_dre("HN:HZ", "revestimento") 4df_dre_consolidado = ler_dre("GT:HF", "consolidado")

Fazendo união de todos os dataframes gerados:

1df_dre = df_dre_madeira.union(df_dre_deca).union(df_dre_revestimento).union(df_dre_consolidado)

4.2.1.2 Indicadores do EVA

Definindo função para Indicadores do EVA, lendo a sheet EVA do arquivo:

1#indicadores de valor agregado: 2def ler_eva(celulas, negocio): 3 df = pd.read_excel(fechamento_gerencial, "EVA", skiprows = 2, nrows = 50, usecols = celulas, index_col = 0).transpose().iloc[1:, 18].reset_index() 4 df.columns = eva_header 5 df = spark.createDataFrame(df) 6 df = df.selectExpr(eva_header[0], "'{}' as negocio".format(negocio), eva_header[1]) 7 df = df.withColumn("mes", when(df.mes.like("JAN%"), "01/01/" + ano_atual) 8 .when(df.mes.like("FEV%"), "01/02/" + ano_atual) 9 .when(df.mes.like("MAR%"), "01/03/" + ano_atual) 10 .when(df.mes.like("ABR%"), "01/04/" + ano_atual) 11 .when(df.mes.like("MAI%"), "01/05/" + ano_atual) 12 .when(df.mes.like("JUN%"), "01/06/" + ano_atual) 13 .when(df.mes.like("JUL%"), "01/07/" + ano_atual) 14 .when(df.mes.like("AGO%"), "01/08/" + ano_atual) 15 .when(df.mes.like("SET%"), "01/09/" + ano_atual) 16 .when(df.mes.like("OUT%"), "01/10/" + ano_atual) 17 .when(df.mes.like("NOV%"), "01/11/" + ano_atual) 18 .when(df.mes.like("DEZ%"), "01/12/" + ano_atual) 19 .otherwise(None)) 20 df = df.withColumn("mes", to_date("mes", "dd/MM/yyyy")) 21 return df

Aplicando a função para cada negócio, passando o parâmetro de range de células onde se encontram os dados:

1df_eva_madeira = ler_eva("GG:GT", "madeira") 2df_eva_deca = ler_eva("BA:BN", "deca") 3df_eva_revestimento = ler_eva("EY:FL", "revestimento") 4df_eva_consolidado = ler_eva("FP:GC", "consolidado")

Fazendo união de todos os dataframes gerados e ajustando a ordem de grandeza do campo eva_recorrente:

1df_eva = df_eva_madeira.union(df_eva_deca).union(df_eva_revestimento).union(df_eva_consolidado).withColumn("eva_recorrente", col("eva_recorrente")/1000)

4.2.2 Arquivo Fluxo_de_Caixa_Oficial_2022.xlsx

4.2.2.1 Indicadores de Fluxo de Caixa

Definindo função para Indicadores de Fluxo de Caixa, lendo a sheet F.C. REAL do arquivo:

1def ler_fluxo_caixa(celulas, negocio): 2 df = pd.read_excel(fluxo_caixa, "F.C. REAL", skiprows = 1, nrows = 86, usecols = celulas, index_col = 0).transpose().iloc[1:, 54].reset_index() 3 df.columns = fc_header 4 df = spark.createDataFrame(df) 5 df = df.selectExpr(fc_header[0], "'{}' as negocio".format(negocio), fc_header[1]) 6 df = df.withColumn("mes", to_date("mes")) 7 return df

Aplicando a função para cada negócio, passando o parâmetro de range de células onde se encontram os dados:

1df_fc_madeira = ler_fluxo_caixa("FN:GA", "madeira") 2df_fc_deca = ler_fluxo_caixa("BM:BZ", "deca") 3df_fc_revestimento = ler_fluxo_caixa("IT:JG", "revestimento") 4df_fc_consolidado = ler_fluxo_caixa("KJ:KW", "consolidado")

Fazendo união de todos os dataframes gerados:

1df_fc = df_fc_madeira.union(df_fc_deca).union(df_fc_revestimento).union(df_fc_consolidado)

4.2.2.2 Indicadores de PMP

Definindo função para Indicadores de PMP, lendo a sheet CGL.ROL do arquivo:

1def ler_pmp(celulas, negocio): 2 df = pd.read_excel(fluxo_caixa, "CGL.ROL", skiprows = 1, nrows = 50, usecols = celulas, index_col = 0).transpose().iloc[1:, 26].reset_index() 3 df.columns = pmp_header 4 df = spark.createDataFrame(df) 5 df = df.selectExpr(pmp_header[0], "'{}' as negocio".format(negocio), pmp_header[1]) 6 df = df.withColumn("mes", to_date("mes")) 7 return df

Aplicando a função para cada negócio, passando o parâmetro de range de células onde se encontram os dados:

1df_pmp_madeira = ler_pmp("DR:EE", "madeira") 2df_pmp_deca = ler_pmp("AU:BH", "deca") 3df_pmp_revestimento = ler_pmp("FZ:GM", "revestimento") 4df_pmp_consolidado = ler_pmp("GO:HB", "consolidado")

Fazendo união de todos os dataframes gerados:

1df_pmp = df_pmp_madeira.union(df_pmp_deca).union(df_pmp_revestimento).union(df_pmp_consolidado)

4.2.3 Arquivo Forecast_3_9_2022.xlsx

4.2.3.1 Indicadores de Forecast

Definindo função para Indicadores de Forecast:

1def ler_forecast(sheet, celulas, negocio): 2 df = pd.read_excel(forecast, sheet, skiprows = 3, nrows = 83, usecols = celulas, index_col = 0).transpose().iloc[:, [6, 47]].reset_index() 3 df.columns = frc_header 4 df = spark.createDataFrame(df) 5 df = df.selectExpr(frc_header[0], "'{}' as negocio".format(negocio), frc_header[1], frc_header[2]) 6 df = df.withColumn("mes", to_date("mes")) 7 return df

Aplicando a função para cada negócio, passando o parâmetro de range de células onde se encontram os dados:

1df_frc_dre_madeira = ler_forecast("DRE_MAD", "A:M", "madeira") 2df_frc_dre_deca = ler_forecast("DRE_DEC", "A:M", "deca") 3df_frc_dre_revestimento = ler_forecast("DRE_REVEST", "A:M", "revestimento") 4df_frc_dre_consolidado = ler_forecast("DRE_CONSOL", "A:M", "consolidado")

Fazendo união de todos os dataframes gerados:

1df_frc = df_frc_dre_madeira.union(df_frc_dre_deca).union(df_frc_dre_revestimento).union(df_frc_dre_consolidado)

4.2.3.2 Indicadores de Forecast EVA

Definindo função para Indicadores de Forecast EVA, lendo a sheet Base EVA ROIC:

1def ler_forecast_eva(negocio): 2 #desinindo constantantes para cada negócio para utilizar posteriormente o iloc[] e localizar os dados necessários 3 #esses valores pra neg são como os negócios são identificados no arquivo 4 if negocio == "madeira": 5 neg = "Mad. Total" 6 elif negocio == "deca": 7 neg = "Deca" 8 elif negocio == "revestimento": 9 neg = "RC" 10 elif negocio == "consolidado": 11 neg = "Consol" 12 df = pd.read_excel(forecast, "Base EVA ROIC", skiprows = 4, nrows = 130, usecols = "B:Q") 13 #buscando a linha em que a coluna 0 é igual ao negócio, coluna 1 igual a "Mês", coluna 2 igual a "RECORRENTE" e coluna 4 igual a "EVA" 14 df = df.where( 15 (df.iloc[:, 0] == neg) & 16 (df.iloc[:, 1] == "Mês") & 17 (df.iloc[:, 2] == "RECORRENTE") & 18 (df.iloc[:, 3] == "EVA") 19 ).dropna() 20 df = df.transpose().iloc[4: ].reset_index() 21 df.columns = frc_eva_header 22 df = spark.createDataFrame(df) 23 df = df.selectExpr(frc_eva_header[0], "'{}' as negocio".format(negocio), frc_eva_header[1]) 24 df = df.withColumn("mes", when(df.mes == 1, "01/01/" + ano_atual) 25 .when(df.mes == 2, "01/02/" + ano_atual) 26 .when(df.mes == 3, "01/03/" + ano_atual) 27 .when(df.mes == 4, "01/04/" + ano_atual) 28 .when(df.mes == 5, "01/05/" + ano_atual) 29 .when(df.mes == 6, "01/06/" + ano_atual) 30 .when(df.mes == 7, "01/07/" + ano_atual) 31 .when(df.mes == 8, "01/08/" + ano_atual) 32 .when(df.mes == 9, "01/09/" + ano_atual) 33 .when(df.mes == 10, "01/10/" + ano_atual) 34 .when(df.mes == 11, "01/11/" + ano_atual) 35 .when(df.mes == 12, "01/12/" + ano_atual) 36 .otherwise(None)) 37 df = df.withColumn("mes", to_date("mes", "dd/MM/yyyy")) 38 return df

Aplicando a função para cada negócio, passando o parâmetro de range de células onde se encontram os dados:

1df_frc_eva_madeira = ler_forecast_eva("madeira") 2df_frc_eva_deca = ler_forecast_eva("deca") 3df_frc_eva_revestimento = ler_forecast_eva("revestimento") 4df_frc_eva_consolidado = ler_forecast_eva("consolidado")

Fazendo união de todos os dataframes gerados e ajustando a ordem de grandeza do campo frc_eva_recorrente:

1df_frc_eva = df_frc_eva_madeira.union(df_frc_eva_deca).union(df_frc_eva_revestimento).union(df_frc_eva_consolidado).withColumn("frc_eva_recorrente", col("frc_eva_recorrente")/1000)

4.2.4 Arquivo FCST_2_10___Fluxo_de_Caixa_Livre_2022.xlsx

4.2.4.1 Indicadores Forecast Fluxo de Caixa

Definindo função para Indicadores de Forecast fluxo de Caixa, lendo a sheet F.C. FCST:

1def ler_fcf(celulas, negocio): 2 df = pd.read_excel(fcf, "F.C. FCST", skiprows = 1, nrows = 87, usecols = celulas, index_col = 0).transpose().iloc[1:, 54].reset_index() 3 df.columns = fcf_header 4 df = spark.createDataFrame(df) 5 df = df.selectExpr(fcf_header[0], "'{}' as negocio".format(negocio), fcf_header[1]) 6 df = df.withColumn("mes", when(df.mes.like("JAN%"), "01/01/" + ano_atual) 7 .when(df.mes.like("FEV%"), "01/02/" + ano_atual) 8 .when(df.mes.like("MAR%"), "01/03/" + ano_atual) 9 .when(df.mes.like("ABR%"), "01/04/" + ano_atual) 10 .when(df.mes.like("MAI%"), "01/05/" + ano_atual) 11 .when(df.mes.like("JUN%"), "01/06/" + ano_atual) 12 .when(df.mes.like("JUL%"), "01/07/" + ano_atual) 13 .when(df.mes.like("AGO%"), "01/08/" + ano_atual) 14 .when(df.mes.like("SET%"), "01/09/" + ano_atual) 15 .when(df.mes.like("OUT%"), "01/10/" + ano_atual) 16 .when(df.mes.like("NOV%"), "01/11/" + ano_atual) 17 .when(df.mes.like("DEZ%"), "01/12/" + ano_atual) 18 .otherwise(None)) 19 df = df.withColumn("mes", to_date("mes", "dd/MM/yyyy")) 20 return df

Aplicando a função para cada negócio, passando o parâmetro de range de células onde se encontram os dados:

1df_fcf_madeira = ler_fcf("FN:GA", "madeira") 2df_fcf_deca = ler_fcf("BM:BZ", "deca") 3df_fcf_revestimento = ler_fcf("HD:HQ", "revestimento") 4df_fcf_consolidado = ler_fcf("IT:JG", "consolidado")

Fazendo união de todos os dataframes gerados:

1df_frc_fc = df_fcf_madeira.union(df_fcf_deca).union(df_fcf_revestimento).union(df_fcf_consolidado)

4.2.5 Arquivo indicadores_poa_2022.csv

Foi feito o upload desse arquivo diretamente no dbfs do Databricks:

1df_poa = spark.read.format("csv").option("delimiter", ";").option("header", "true").load("dbfs:/FileStore/shared_uploads/arquivos_financeiros/indicadores_poa_2022.csv")

Tratando alguns tipos de campos e ajustando a ordem de grandeza de outro:

1df_poa = df_poa.select( 2 "mes", 3 "negocio", 4 col("poa_receita_liquida_vendas").cast(DoubleType()), 5 col("poa_ebitda_recorrente").cast(DoubleType()), 6 col("poa_eva_recorrente").cast(DoubleType()), 7 col("poa_fluxo_caixa_livre_total").cast(DoubleType()) 8).withColumn("poa_eva_recorrente", col("poa_eva_recorrente")/1000).withColumn("poa_fluxo_caixa_livre_total", col("poa_fluxo_caixa_livre_total")/1000)

4.2.6 Base Final

Todos os dataframes gerados nos passos anteriores agora são unidos para formar um único, utilizando os campos mes e negocio:

1campos_join = ["mes", "negocio"] 2df_indicadores_financeiros = df_dre.join( 3 df_eva, campos_join 4).join( 5 df_fc, campos_join 6).join( 7 df_pmp, campos_join 8).join( 9 df_frc, campos_join 10).join( 11 df_frc_eva, campos_join 12).join( 13 df_frc_fc, campos_join 14).join( 15 df_poa, campos_join 16).orderBy("negocio", "mes")

Sobrescrevendo a tabela final no database :

1df_indicadores_financeiros.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_financeiros_{}".format(ano_atual))

4.3 Base Final:

col_name

data_type

mes

date

negocio

string

receita_liquida_vendas

double

ebitda_recorrente

double

eva_recorrente

double

fluxo_caixa_livre_total

double

pmp

double

frc_receita_liquida_vendas

double

frc_ebitda_recorrente

double

frc_eva_recorrente

double

frc_fluxo_caixa_livre_total

double

poa_receita_liquida_vendas

double

poa_ebitda_recorrente

double

poa_eva_recorrente

double

poa_fluxo_caixa_livre_total

double


5. Indicadores Gente

Indicadores de Recursos Humanos.

  • Notebook: indicadores_gente

  • Job: inteligencia-mercado_job_prod_indicadores_gente

  • Schedule: sextas-feiras, 14:00h

  • Base fim: indicadores_mercado.tb_indicadores_gente

5.1 Origem:

O arquivo indicadores_gente.csv é disponibilizado e atualizado mensalmente no Sharepoint.

5.2 Transformação:

Definindo função para o download do arquivo para o dbfs:

1def download_arquivo(arquivo_download, nome_arquivo): 2 ctx = ClientContext(site_sharepoint).with_credentials(UserCredential(usuario, senha)) 3 web = ctx.load(ctx.web).execute_query() 4 response = File.open_binary(ctx, arquivo_download) 5 response.raise_for_status() 6 with open("/dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/" + nome_arquivo, "wb") as pasta: 7 pasta.write(response.content)

Lendo o arquivo:

1df_gente = spark.read.format("csv").option("delimiter", ";").option("header", "true").load("dbfs:/FileStore/shared_uploads/arquivos_diversos_cognitivo/indicadores_gente.csv")

Aplicando regra de negócio para o campo taxa_afastamento:

1df_gente = df_gente.withColumn("taxa_afastamento", col("taxa_absenteismo_com_afastamento") - col("taxa_absenteismo_sem_afastamento"))

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_gente.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_gente")

5.3 Base Final:

col_name

data_type

mes

date

negocio

string

quantidade_cargos_lideranca_mulheres

int

quantidade_cargos_lideranca_total

int

taxa_desligamentos_voluntarios

double

taxa_desligamentos_involuntarios

double

taxa_desligamentos_total

double

taxa_absenteismo_sem_afastamento

double

taxa_absenteismo_com_afastamento

double

taxa_afastamento

double


7. Indicadores Margem Madeira

Indicadores de Margem para Madeira.

  • Notebook: indicadores_margem_madeira

  • Job: inteligencia-mercado_job_prod_indicadores_margem_madeira

  • Schedule: diário/14:22h

  • Base fim: indicadores_mercado.tb_indicadores_margem_madeira

7.1 Origem:

É feita uma consulta na fonte de dados vw_ren_rateio_aj_SQL, no site Madeira do Tableau Server:

7.2 Transformação:

São utilizadas as bibliotecas tableauserverclient e tableauhyperapi do Python:

1import tableauserverclient as tsc 2import tableauhyperapi 3from tableauhyperapi import HyperProcess, Telemetry, Connection, TableName, escape_name, escape_string_literal

Definindo variáveis para conexão no Tableau Server:

1usuario = "***" 2senha = "***" 3site = "Madeira" 4servidor = "https://analytics.duratex.com.br" 5id_datasource = "e88f21c1-bb2f-4516-b612-b62e8af74b94" 6nome_database = "vw_ren_rateio_aj_SQL" 7diretorio_download = "/dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/"

Definindo parâmetros da conexão com o Tableau Server:

1autenticador = tsc.TableauAuth(usuario, senha, site) 2servidor = tsc.Server(servidor)

Fazendo o download do arquivo .tdsx:

1with servidor.auth.sign_in(autenticador): 2 caminho_arquivo = servidor.datasources.download(id_datasource, filepath = diretorio_download, include_extract = True)

O arquivo de extensão .tdsx nada mais é que uma compactação dos arquivos de fonte de dados do Tableau. Portanto foi utilizada a biblioteca zipfile do Python para fazer essa descompactação em diretório do dbfs:

1with zipfile.ZipFile(diretorio_download + nome_database + ".tdsx", "r") as arquivo_zipado: 2 arquivo_zipado.extractall(diretorio_download)

Após a descompactação, é utilizada a biblioteca tableauhyperapi para extrair os dados do arquivo .hyper, que foi descompactado no diretório /dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/Data/Extracts/ no processo anterior. É feita uma query para trazer apenas os campos necessários:

1with HyperProcess(telemetry = Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU) as hyper: 2 with Connection(endpoint = hyper.endpoint, database = diretorio_download + "Data/Extracts/" + nome_database + ".hyper") as conexao: 3 with conexao.execute_query(query = f"select {escape_name('cd_competencia')}, {escape_name('dc_linha_produto')}, {escape_name('gerencia')}, {escape_name('vl_receita_aj')}, {escape_name('md_m3_quantidade')}, {escape_name('vl_custo_industrial_total_CIT')}, {escape_name('vl_custo_comercial_total_CCO')} from {TableName('Extract', 'Extract')} where {escape_name('cd_setor_atividade')} = {escape_string_literal('CH')} and {escape_name('cd_competencia')} >= 202101 and {escape_name('gerencia')} = {escape_string_literal('MERCADO EXTERNO')}") as resultado: 4 resultado_consulta = list(resultado)

Definindo schema para o dataframe que será criado a partir da lista resultado_consulta:

1esquema_margem = StructType([ 2 StructField('mes', StringType(), True), 3 StructField('produto', StringType(), True), 4 StructField('mercado', StringType(), True), 5 StructField('receita', DoubleType(), True), 6 StructField('quantidade', DoubleType(), True), 7 StructField('valor_cit', DoubleType(), True), 8 StructField('valor_cco', DoubleType(), True) 9])

Criando o dataframe:

1df_margem = spark.createDataFrame(resultado_consulta, esquema_margem)

Utilizando o Spark SQL para aplicar algumas regras de negócio:

1df_margem_calculos = spark.sql(""" 2 select 3 mes, 4 (case when produto like 'MDF%' then 'mdf' when produto like 'MDP%' then 'mdp' else 'outros' end) as produto, 5 (sum(receita) / sum(quantidade)) as receita_m3, 6 (sum(valor_cit) / sum(quantidade)) as cit_m3, 7 (sum(valor_cco) / sum(quantidade)) as cco_m3 8 9 from margem_view 10 11 group by 1, 2 12 13 order by 1, 2 asc 14""")

Aplicando mais algumas regras de negócio:

1df_margem_calculos = df_margem_calculos.withColumn("mop_m3", col("receita_m3") - col("cit_m3") - col("cco_m3")).withColumn("percentual_margem", col("mop_m3") / col("receita_m3"))

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_margem_calculos.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_margem_madeira")

7.3 Base Final:

col_name

data_type

mes

string

produto

string

receita_m3

double

cit_m3

double

cco_m3

double

mop_m3

double

9. Indicadores OEE Madeira

Indicadores de OEE para Madeira.

  • Notebook: indicadores_oee_madeira

  • Job: inteligencia-mercado_job_prod_indicadores_oee_madeira

  • Schedule: diário/12:00h

  • Base fim: indicadores_mercado.tb_indicadores_oee_madeira

9.1 Origem:

O arquivo indicadores_oee_madeira.xlsx é disponibilizado e atualizado mensalmente no Sharepoint.

9.2 Transformação:

Definindo função para ler o arquivo:

1def extrair_excel(arquivo_leitura, nome_sheet, celulas, colunas, pular_linhas): 2 ctx = ClientContext(site_sharepoint).with_credentials(UserCredential(usuario, senha)) 3 web = ctx.load(ctx.web).execute_query() 4 response = File.open_binary(ctx, arquivo_leitura) 5 response.raise_for_status() 6 df = pd.read_excel(response.content, skiprows = pular_linhas, usecols = celulas, sheet_name = nome_sheet, names = colunas) 7 df = spark.createDataFrame(df) 8 return df

Lendo a sheet cru do arquivo, definindo os parâmetros de range de células, nome das células e quantidade de linhas a serem desconsideradas. Utilizando o while e a função isinstance() para repetir o processo até que o dataframe esteja criado:

1df_oee_cru = [] 2while not isinstance(df_oee_cru, DataFrame): 3 try: 4 df_oee_cru = extrair_excel(arquivo, "cru", "B:D", ["mes", "oee", "oee_meta_global"], 0) 5 except: 6 pass

Definindo o nome do produto como constante:

1df_oee_cru = df_oee_cru.withColumn("produto", lit("cru"))

Lendo a sheet revestido do arquivo, definindo os parâmetros de range de células, nome das células e quantidade de linhas a serem desconsideradas:

1df_oee_revestido = [] 2while not isinstance(df_oee_revestido, DataFrame): 3 try: 4 df_oee_revestido = extrair_excel(arquivo, "revestido", "B:D", ["mes", "oee", "oee_meta_global"], 0) 5 except: 6 pass

Definindo o nome do produto como constante:

1df_oee_revestido = df_oee_revestido.withColumn("produto", lit("revestido"))

Lendo a sheet historico do arquivo, definindo os parâmetros de range de células, nome das células e quantidade de linhas a serem desconsideradas:

1df_oee_historico = [] 2while not isinstance(df_oee_historico, DataFrame): 3 try: 4 df_oee_historico = extrair_excel(arquivo, "historico", "B:E", ["mes", "oee", "oee_meta_global", "produto"], 0) 5 except: 6 pass

Fazendo a união de todos os dataframes criados:

1df_oee_madeira = df_oee_historico.union(df_oee_cru).union(df_oee_revestido)

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_oee_madeira.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_oee_madeira")

9.3 Base Final:

col_name

data_type

mes

date

oee

double

oee_meta_global

double

produto

string

10. Indicadores OTIF Deca

Indicadores de OTIF para Deca.

  • Notebook: indicadores_otif_deca

  • Job: inteligencia-mercado_job_prod_indicadores_otif_deca

  • Schedule: diário/12:00h

  • Base fim: indicadores_mercado.tb_indicadores_otif_deca

10.1 Origem:

O arquivo dados_otif_deca.csv é disponibilizado e atualizado mensalmente no Sharepoint.

10.2 Transformação:

Definindo função para fazer o download do arquivo em diretório do dbfs:

1def download_arquivo(arquivo_download, nome_arquivo): 2 ctx = ClientContext(site_sharepoint).with_credentials(UserCredential(usuario, senha)) 3 web = ctx.load(ctx.web).execute_query() 4 response = File.open_binary(ctx, arquivo_download) 5 response.raise_for_status() 6 with open("/dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/" + nome_arquivo, "wb") as pasta: 7 pasta.write(response.content)

Lendo arquivo:

1df_otif_deca = spark.read.format("csv").option("delimiter", ";").option("header", "true").load("dbfs:/FileStore/shared_uploads/arquivos_diversos_cognitivo/dados_otif_deca.csv")

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_otif_deca.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_otif_deca")

10.3 Base Final:

col_name

data_type

mes

date

segmento_deca

string

otif_deca

double

meta_otif_deca

double

11. Indicadores OTIF Madeira

Indicadores de OTIF para Madeira.

  • Notebook: indicadores_otif_madeira

  • Job: inteligencia-mercado_job_prod_indicadores_otif_madeira

  • Schedule: diário/12:00h

  • Base fim: indicadores_mercado.tb_indicadores_otif_madeira

11.1 Origem:

O arquivo dados_otif_madeira.csv é disponibilizado e atualizado mensalmente no Sharepoint.

11.2 Transformação:

Definindo função para fazer o download do arquivo em diretório do dbfs:

1def download_arquivo(arquivo_download, nome_arquivo): 2 ctx = ClientContext(site_sharepoint).with_credentials(UserCredential(usuario, senha)) 3 web = ctx.load(ctx.web).execute_query() 4 response = File.open_binary(ctx, arquivo_download) 5 response.raise_for_status() 6 with open("/dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/" + nome_arquivo, "wb") as pasta: 7 pasta.write(response.content)

Lendo arquivo:

1df_otif_madeira = spark.read.format("csv").option("delimiter", ";").option("header", "true").load("dbfs:/FileStore/shared_uploads/arquivos_diversos_cognitivo/dados_otif_madeira.csv")

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_otif_madeira.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_otif_madeira")

11.3 Base Final:

col_name

data_type

mes

date

segmento_madeira

string

otif_madeira

double

meta_otif_madeira

double

12. Indicadores Panorama Mercado Madeira

Indicadores de Panorama de Mercado para Madeira.

  • Notebook: indicadores_panorama_mercado_madeira

  • Job: inteligencia-mercado_job_prod_indicadores_panorama_mercado_madeira

  • Schedule: diário/13:00h

  • Base fim: indicadores_mercado.tb_indicadores_panorama_mercado_madeira

12.1 Origem:

Database DEV do Redshift (dtx-deca-sellin.czcbob9woqfg.us-east-1.redshift.amazonaws.com), schema madeira e tabela vw_mercado_panorama_v2.

12.2 Transformação:

É feita uma query da tabela já aplicando algumas regras de negócio:

1query = "select competencia, mercado, case when produto_detalhe in('MDF Fino Cru', 'MDF Grosso Cru') then 'MDF Cru' when produto_detalhe in('MDF Fino BP', 'MDF Grosso BP') then 'MDF Revestido' when produto_detalhe = 'MDP Cru' then 'MDP Cru' when produto_detalhe = 'MDP BP' then 'MDP Revestido' end as produto, sum(volume_m3) as volume_m3_total from madeira.vw_mercado_panorama_v2 where competencia >= 202101 and mercado = 'MI' group by 1, 2, 3 order by 1"

Renomeando campos:

1colunas_panorama_mercado = ["mes", "mercado", "produto", "volume_m3_total"] 2df_panorama_mercado = panorama_mercado.toDF(*colunas_panorama_mercado)

Ajustando campo mes:

1df_panorama_mercado = df_panorama_mercado.withColumn("mes", col("mes").cast(StringType())).withColumn("mes", concat(substring(col("mes"), 1, 4), lit("-"), substring(col("mes"), 5, 2), lit("-"), lit("01"))).withColumn("mes", col("mes").cast(DateType()))

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_panorama_mercado.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_panorama_mercado_madeira")

12.3 Base Final:

col_name

data_type

mes

date

mercado

string

produto

string

volume_m3_total

double

13. Indicadores Produtividade Deca

Indicadores de Produtividade para Deca.

  • Notebook: indicadores_produtividade_deca

  • Job: inteligencia-mercado_job_prod_indicadores_produtividade_deca

  • Schedule: diário/12:00h

  • Base fim: indicadores_mercado.tb_indicadores_minuto_homem_deca

13.1 Origem:

O arquivo Produtividade DECA&RC.xlsx é disponibilizado e atualizado mensalmente no Sharepoint.

13.2 Transformação:

Definindo função para fazer o download do arquivo para diretório do dbfs:

1def download_arquivo(arquivo_download, nome_arquivo): 2 ctx = ClientContext(site_sharepoint).with_credentials(UserCredential(usuario, senha)) 3 web = ctx.load(ctx.web).execute_query() 4 response = File.open_binary(ctx, arquivo_download) 5 with open("/dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/" + nome_arquivo, "wb") as pasta: 6 pasta.write(response.content)

Fazendo o download do arquivo:

1download_arquivo("/personal/henrique_fantin_duratex_com_br/Documents/Min_HEq/Produtividade%20DECA%26RC.xlsx", "Produtividade_DECA_RC.xlsx")

Lendo o arquivo:

1produtividade_deca = pd.ExcelFile("/dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/Produtividade_DECA_RC.xlsx")

Definindo função para ler os dados dos diferentes segmentos Deca. Passando os parâmetros de negócio, range de células, quantidade de linhas desconsideradas e quantidade de linhas consideradas, a função gera dois dataframes, um para produtividade e outro para POA. Isso é feito através de filtro na coluna: quando é diferente de POA, então é produtividade.

1def ler_produtividade(negocio, celulas, pular_linhas, numero_linhas): 2 #lendo arquivo excel 3 df = pd.read_excel(produtividade_deca, skiprows = pular_linhas, nrows = numero_linhas, usecols = celulas, index_col = 0).transpose().reset_index() 4 #definindo dataframe para o poa 5 df_poa = df 6 #definindo dataframe para produtividade, coluna 1 é diferente de "POA" 7 df = df.where(df.iloc[:, 1] != "POA") 8 df = df.iloc[:, [0, 1, numero_linhas]] 9 df.columns = ["mes", "ano", "produtividade"] 10 df = df.dropna(axis = 0) 11 df = spark.createDataFrame(df) 12 df = df.withColumn("ano", col("ano").cast(IntegerType()).cast(StringType())) 13 #ajustandando campo mes para produtividade 14 df = df.withColumn("mes", when(col("mes").like("JAN%"), concat(lit("01/01/"), col("ano"))) 15 .when(col("mes").like("FEV%"), concat(lit("01/02/"), col("ano"))) 16 .when(col("mes").like("MAR%"), concat(lit("01/03/"), col("ano"))) 17 .when(col("mes").like("ABR%"), concat(lit("01/04/"), col("ano"))) 18 .when(col("mes").like("MAI%"), concat(lit("01/05/"), col("ano"))) 19 .when(col("mes").like("JUN%"), concat(lit("01/06/"), col("ano"))) 20 .when(col("mes").like("JUL%"), concat(lit("01/07/"), col("ano"))) 21 .when(col("mes").like("AGO%"), concat(lit("01/08/"), col("ano"))) 22 .when(col("mes").like("SET%"), concat(lit("01/09/"), col("ano"))) 23 .when(col("mes").like("OUT%"), concat(lit("01/10/"), col("ano"))) 24 .when(col("mes").like("NOV%"), concat(lit("01/11/"), col("ano"))) 25 .when(col("mes").like("DEZ%"), concat(lit("01/12/"), col("ano"))) 26 .otherwise(None)) 27 df = df.withColumn("mes", to_date("mes", "dd/MM/yyyy")) 28 #trabalhando com dataframe para poa, quando coluna 1 é igual a "POA" 29 df_poa = df_poa.where(df_poa.iloc[:, 1] == "POA") 30 df_poa = df_poa.iloc[:, [0, numero_linhas]] 31 df_poa .columns = ["mes", "poa_produtividade"] 32 df_poa = df_poa.dropna(axis = 0) 33 df_poa = spark.createDataFrame(df_poa) 34 #ajustando campo mes 35 df_poa = df_poa.withColumn("mes", when(col("mes").like("JAN%"), "01/01/" + ano_atual) 36 .when(col("mes").like("FEV%"), "01/02/" + ano_atual) 37 .when(col("mes").like("MAR%"), "01/03/" + ano_atual) 38 .when(col("mes").like("ABR%"), "01/04/" + ano_atual) 39 .when(col("mes").like("MAI%"), "01/05/" + ano_atual) 40 .when(col("mes").like("JUN%"), "01/06/" + ano_atual) 41 .when(col("mes").like("JUL%"), "01/07/" + ano_atual) 42 .when(col("mes").like("AGO%"), "01/08/" + ano_atual) 43 .when(col("mes").like("SET%"), "01/09/" + ano_atual) 44 .when(col("mes").like("OUT%"), "01/10/" + ano_atual) 45 .when(col("mes").like("NOV%"), "01/11/" + ano_atual) 46 .when(col("mes").like("DEZ%"), "01/12/" + ano_atual) 47 .otherwise(None)) 48 df_poa = df_poa.withColumn("mes", to_date("mes", "dd/MM/yyyy")) 49 #definindo dataframe final, fazendo um join com os anteriores 50 df_final = df_poa.join(df, df_poa.mes == df.mes, "left").drop(df.mes) 51 df_final = df_final.withColumn("segmento", lit(negocio)) 52 df_final = df_final.select("mes", "segmento", "poa_produtividade", "produtividade") 53 df_final = df_final.withColumn("poa_produtividade", round(col("poa_produtividade"), 2)).withColumn("produtividade", round(col("produtividade"), 2)) 54 return df_final

Aplicando função para metais, já definindo os parâmetros de range de células, quantidade de linhas ignoradas e quantidade de linhas consideradas:

1df_metais = ler_produtividade("metais", "FA:FZ", 21, 5)

Aplicando função para loucas:

1df_loucas = ler_produtividade("loucas", "FA:FZ", 54, 8)

Aplicando função para Hydra:

1df_hydra = ler_produtividade("hydra", "FA:FZ", 93, 4)

Aplicando função para revestimento:

1df_revestimento = ler_produtividade("revestimento", "FA:FZ", 145, 7)

Fazendo a união de todos os dataframes criados:

1df_produtividade_deca = df_metais.union(df_loucas).union(df_hydra).union(df_revestimento)

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_produtividade_deca.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_minuto_homem_deca")

13.3 Base Final:

col_name

data_type

mes

date

segmento

string

poa_produtividade

double

produtividade

double

14. Indicadores SAC

Indicadores de SAC.

  • Notebook: indicadores_sac

  • Job: inteligencia-mercado_job_prod_indicadores_sac

  • Schedule: diário/12:00h

  • Base fim: indicadores_mercado.tb_indicadores_sac

14.1 Origem:

O arquivo dados_sac.csv é disponibilizado e atualizado mensalmente no Sharepoint.

14.2 Transformação:

Definindo função para fazer o download do arquivo em diretório do dbfs:

1def download_arquivo(arquivo_download, nome_arquivo): 2 ctx = ClientContext(site_sharepoint).with_credentials(UserCredential(usuario, senha)) 3 web = ctx.load(ctx.web).execute_query() 4 response = File.open_binary(ctx, arquivo_download) 5 response.raise_for_status() 6 with open("/dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/" + nome_arquivo, "wb") as pasta: 7 pasta.write(response.content)

Lendo arquivo:

1df_sac = spark.read.format("csv").option("delimiter", ";").option("header", "true").load("dbfs:/FileStore/shared_uploads/arquivos_diversos_cognitivo/dados_sac.csv")

Aplicando regra de negócio para o campo saldo:

1df_sac = df_sac.withColumn("saldo", col("casos_abertos") - col("casos_fechados"))

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_sac.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_sac")

14.3 Base Final:

col_name

data_type

mes

date

visao

string

negocio

string

casos_abertos

int

casos_fechados

int

saldo

int

15. Indicadores Sell-in Deca - Chuveiros

Indicadores de Sell-in para Deca.

  • Notebook: indicadores_sell_in_deca_chuveiros

  • Base fim: indicadores_mercado.tb_indicadores_sell_in_deca_chuveiros

15.1 Origem:

O arquivo foi disponibilizado e foi feito seu upload direto no dbfs do Databricks. Motivo é que esses dados não são atualizáveis. Diretório de upload: /dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/sell_in_deca_chuveiros.xlsx.

15.2 Transformação:

Lendo sheet do arquivo:

1arquivo_sell_in = pd.ExcelFile("/dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/sell_in_deca_chuveiros.xlsx") 2df_sell_in_deca = pd.read_excel(arquivo_sell_in, "Banco de dados - Construcheck -")

Renomeando colunas:

1df_sell_in_deca.columns = ["marca", "canal_abastecimento", "regiao", "macro_categoria", "material", "tipo", "sku", "meses", "desviador", "linha", "tensao", "potencia", "modelo", "ean", "valor", "volume"]

Tratando os valores missing:

1objs = df_sell_in_deca.select_dtypes(include = "object").columns 2df_sell_in_deca[objs] = df_sell_in_deca[objs].fillna("").astype(str)

Transformando em dataframe do Spark:

1df_sell_in_deca = spark.createDataFrame(df_sell_in_deca)

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_sell_in_deca.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_sell_in_deca_chuveiros")

15.3 Base Final:

col_name

data_type

marca

string

canal_abastecimento

string

regiao

string

macro_categoria

string

material

string

tipo

string

sku

string

meses

timestamp

desviador

string

linha

string

tensao

string

potencia

string

modelo

string

ean

string

valor

double

volume

double

16. Indicadores Sell-in Deca - Cubas

Indicadores de Sell-in para Deca.

  • Notebook: indicadores_sell_in_deca_cubas

  • Base fim: indicadores_mercado.tb_indicadores_sell_in_deca_cubas

16.1 Origem:

O arquivo foi disponibilizado e foi feito seu upload direto no dbfs do Databricks. Motivo é que esses dados não são atualizáveis. Diretório de upload: /dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/sell_in_deca_cubas.xlsx.

16.2 Transformação:

Lendo sheet do arquivo:

1arquivo_sell_in = pd.ExcelFile("/dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/sell_in_deca_cubas.xlsx") 2df_sell_in_deca = pd.read_excel(arquivo_sell_in, "Banco de dados - Construcheck -")

Renomeando colunas:

1df_sell_in_deca.columns = ["marca", "canal_abastecimento", "regiao", "macro_categoria", "material", "aplicacao", "instalacao", "sku", "meses", "formato", "cor", "acabamento", "ean", "valor", "volume"]

Tratando os valores missing:

1objs = df_sell_in_deca.select_dtypes(include = "object").columns 2df_sell_in_deca[objs] = df_sell_in_deca[objs].fillna("").astype(str)

Transformando em dataframe do Spark:

1df_sell_in_deca = spark.createDataFrame(df_sell_in_deca)

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_sell_in_deca.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_sell_in_deca_cubas")

16.3 Base Final:

col_name

data_type

marca

string

canal_abastecimento

string

regiao

string

macro_categoria

string

material

string

aplicacao

string

instalacao

string

sku

string

meses

timestamp

formato

string

cor

string

acabamento

string

ean

string

valor

double

volume

double

17. Indicadores Sell-in Deca - Torneiras

Indicadores de Sell-in para Deca.

  • Notebook: indicadores_sell_in_deca_torneiras

  • Base fim: indicadores_mercado.tb_indicadores_sell_in_deca_torneiras

17.1 Origem:

O arquivo foi disponibilizado e foi feito seu upload direto no dbfs do Databricks. Motivo é que esses dados não são atualizáveis. Diretório de upload: /dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/sell_in_deca_torneiras.xlsx.

17.2 Transformação:

Lendo sheet do arquivo:

1arquivo_sell_in = pd.ExcelFile("/dbfs/FileStore/shared_uploads/arquivos_diversos_cognitivo/sell_in_deca_torneiras.xlsx") 2df_sell_in_deca = pd.read_excel(arquivo_sell_in, "Banco de dados - Construcheck -")

Renomeando colunas:

1df_sell_in_deca.columns = ["marca", "canal_abastecimento", "regiao", "macro_categoria", "material", "subcategoria", "aplicacao", "instalacao", "sku", "meses", "bica", "linha", "tensao", "potencia", "modelo", "ean", "valor", "volume"]

Tratando os valores missing:

1objs = df_sell_in_deca.select_dtypes(include = "object").columns 2df_sell_in_deca[objs] = df_sell_in_deca[objs].fillna("").astype(str)

Transformando em dataframe do Spark:

1df_sell_in_deca = spark.createDataFrame(df_sell_in_deca)

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_sell_in_deca.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_sell_in_deca_torneiras")

17.3 Base Final:

col_name

data_type

marca

string

canal_abastecimento

string

regiao

string

macro_categoria

string

material

string

subcategoria

string

aplicacao

string

instalacao

string

sku

string

meses

timestamp

bica

string

linha

string

tensao

string

potencia

string

modelo

string

ean

string

valor

double

volume

double

18. Indicadores Market Share, Sell-in e Sell-out Madeira

Indicadores para Market Share, Sell-in e Sell-out de Madeira.

  • Notebook: indicadores_share_sell_in_sell_out_madeira

  • Job: inteligencia-mercado_job_prod_indicadores_share_sell_in_sell_out_madeira

  • Schedule: diário/12:00h

  • Base fim: indicadores_mercado.tb_indicadores_share_sell_in_sell_out_madeira

18.1 Origem:

Database DEV do Redshift (dtx-deca-sellin.czcbob9woqfg.us-east-1.redshift.amazonaws.com), schema madeira e view vw_base_mktshare_sellin_sellout.

18.2 Transformação:

Fazendo uma query com um select de todos os campos da tabela origem:

1query = "select * from madeira.vw_base_mktshare_sellin_sellout" 2multiple_run_parameters = dbutils.notebook.entry_point.getCurrentBindings() 3bucket_name = multiple_run_parameters["bucket_name"] 4consulta_madeira = redshift_to_dataframe(query = query, filename = "vw_base_mktshare_sellin_sellout", bucket_name = bucket_name)

Renomeando as colunas:

1olunas_madeira = ["tipo", "ano", "competencia", "mercado", "produto", "produto_detalhe", "volume_m3", "volume_m2", "volume_m3_cap_dtx", "volume_m3_cap_mercex", "data_atualizacao_base", "produto_segmento", "quantidade_volume_m3_liquido", "quantidade_volume_m2"] 2df_madeira = consulta_madeira.toDF(*colunas_madeira)

Ajustando tipos dos campos:

1df_madeira = df_madeira.withColumn("ano", col("ano").cast(IntegerType())).withColumn("competencia", col("competencia").cast(IntegerType())).withColumn("volume_m3", col("volume_m3").cast(DoubleType())).withColumn("volume_m2", col("volume_m2").cast(DoubleType())).withColumn("volume_m3_cap_dtx", col("volume_m3_cap_dtx").cast(DoubleType())).withColumn("volume_m3_cap_mercex", col("volume_m3_cap_mercex").cast(DoubleType()))

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_madeira.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_share_sell_in_sell_out_madeira")

18.3 Base Final:

col_name

data_type

tipo

string

ano

int

competencia

int

mercado

string

produto

string

produto_detalhe

string

volume_m3

double

volume_m2

double

volume_m3_cap_dtx

double

volume_m3_cap_mercex

double

data_atualizacao_base

date

produto_segmento

string

quantidade_volume_m3_liquido

double

quantidade_volume_m2

double

19. Indicadores Vendas Deca

Indicadores para Vendas de Deca.

  • Notebook: indicadores_venda_deca

  • Job: inteligencia-mercado_job_prod_indicadores_venda_deca

  • Schedule: diário/10:00h

  • Base fim: indicadores_mercado.tb_indicadores_vendas_deca

19.1 Origem:

Consulta da tabela tb_resultado_comercial da database large e consulta database DEV do Redshift (dtx-deca-sellin.czcbob9woqfg.us-east-1.redshift.amazonaws.com), schema large e tabela tb_metas_comercial_hierarquia_produto.

19.2 Transformação:

Definindo função para criação de dataframe com range de datas, desde 01/01/2022 até hoje.

1for negocio in negocios: 2 df = pd.date_range(start = "2022-01-01", end = date.today()) 3 df = pd.DataFrame(df, columns = ["data"]) 4 df = spark.createDataFrame(df) 5 df = df.withColumn("data", to_date("data", 'yyyy-MM-dd')).withColumn("mes", to_date(date_format("data", 'yyyy-MM-01'))) 6 df = df.withColumn("negocio", lit(negocio)) 7 if negocio == negocios[0]: 8 df_datas = df 9 else: 10 df_datas = df_datas.union(df)

Sobrescrevendo a tabela de datas na database indicadores_mercado, para ser utilizada em outra etapa do processo:

1df_datas.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_datas")

Fazendo consulta para vendas na large, já aplicando algumas regras de negócio:

1df_vendas = spark.sql(""" 2 select 3 t1.data_competencia as data, 4 to_date(date_format(data_competencia, 'yyyy-MM-01')) as mes, 5 (case when t1.codigo_setor_atividade = 'CS' then 'loucas' 6 when t1.codigo_setor_atividade = 'MS' then 'metais' 7 when t1.codigo_setor_atividade = '01' then 'revestimento' 8 when t1.codigo_setor_atividade = 'HY' then 'hydra' end) as negocio, 9 round(sum(t1.valor_receita_liquida), 2) as receita_liquida_vendas 10 from large.tb_resultado_comercial t1 11 where t1.codigo_setor_atividade in("CS", "MS", "HY", "01") 12 and t1.status_ordem_venda in("EXPORTAÇÃO", "VENDA", "DEVOLUÇÃO", "CANCELAMENTO", "CINI") 13 and t1.data_competencia between '2022-01-01' and (current_date() - 1) 14 group by 1, 2, 3 15 order by 3, 2, 1 asc 16""")

Fazendo join entre o dataframe de datas e o de vendas, com o objetivo de agregar todos os dias do range na base, mesmo que sem registro de vendas:

1df_vendas_ajustado = df_datas.join(df_vendas, ["data", "mes", "negocio"], "left")

Definindo um particionamento no dataframe df_vendas_ajustado, com o objetivo de posteriormente fazer uma soma acumuladas das vendas agrupada por negócio e mês:

1part = (Window.partitionBy("negocio", "mes").orderBy("data").rangeBetween(Window.unboundedPreceding, 0))

Criando campo com soma acumulada, utilizando a partição criada anteriormente:

1df_vendas_acumuladas = df_vendas_ajustado.withColumn("receita_liquida_vendas_acumuladas", functions.sum("receita_liquida_vendas").over(part))

Criando campo com soma acumulada, utilizando a partição criada anteriormente:

1df_vendas_acumuladas = df_vendas_ajustado.withColumn("receita_liquida_vendas_acumuladas", functions.sum("receita_liquida_vendas").over(part))

Sobrescrevendo a tabela de datas na database indicadores_mercado, para ser utilizada em outra etapa do processo:

1df_vendas_acumuladas.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_vendas_diarias_deca")

Então, é feita uma query, já aplicando algumas regras de negócio, na large do Redshift, com o objetivo de obter as metas:

1query = "select data_competencia as mes, (case when codigo_setor_atividade = 'CS' then 'loucas' when codigo_setor_atividade = 'MS' then 'metais' when codigo_setor_atividade = '01' then 'revestimento' when codigo_setor_atividade = 'HY' then 'hydra' end) as negocio, tipo_meta, sum(valor_receita_liquida) as valor_meta from large.tb_metas_comercial_hierarquia_produto where data_competencia >= '2021-01-01' and codigo_setor_atividade in('CS', 'HY', 'MS', '01') and tipo_meta in('POA', 'PEV') group by 1, 2, 3 order by 1, 2, 3" 2multiple_run_parameters = dbutils.notebook.entry_point.getCurrentBindings() 3bucket_name = multiple_run_parameters["bucket_name"] 4consulta_metas = redshift_to_dataframe(query = query, filename = "tb_metas_comercial_hierarquia_produto", bucket_name = bucket_name)

Renomeando colunas:

1colunas_metas = ["mes", "negocio", "tipo_meta", "valor_meta_mes"] 2df_metas = consulta_metas.toDF(*colunas_metas)

Próximo passo é definir uma função para cálculo de quantidade de dias dentro de cada mês, com o objetivo de posteriormente calcular a meta diárias através da mensal:

1def numero_dias(data): 2 mes = data.month 3 ano = data.year 4 #tratativa caso fevereiro (ano bissexto) 5 cons = 0 6 #todo ano divisível por 400 e ao mesmo tempo 100 é bissexto 7 if ano % 400 == 0 and ano % 100 == 0: 8 cons = 1 9 #todo ano divisível por 4 é bissexto 10 elif ano % 4 == 0: 11 cons = 1 12 else: 13 cons = 0 14 if mes == 2: 15 return 28 + cons 16 #meses com quantidade ímpar de dias 17 impares = [1, 3, 5, 7, 8, 10, 12] 18 if mes in impares: 19 return 31 20 return 30

Aplicando a função e calculando a meta diária:

1numero_dias_udf = udf(lambda z: numero_dias(z), IntegerType()) 2df_metas = df_metas.withColumn("valor_meta_diaria", (col("valor_meta_mes") / col("numero_dias_mes")))

Sobrescrevendo a tabela de datas na database indicadores_mercado, para ser utilizada em outra etapa do processo:

1df_metas.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_metas_mensais_deca")

Criando colunas para cada tipo de meta, PEV e POA:

1df_metas_pev = df_metas.where(col("tipo_meta") == "PEV").withColumnRenamed("valor_meta_diaria", "valor_meta_diaria_pev").drop("tipo_meta", "valor_meta_mes", "numero_dias_mes").withColumn("valor_meta_diaria_pev", functions.round("valor_meta_diaria_pev", 2)) 2df_metas_poa = df_metas.where(col("tipo_meta") == "POA").withColumnRenamed("valor_meta_diaria", "valor_meta_diaria_poa").drop("tipo_meta", "valor_meta_mes", "numero_dias_mes").withColumn("valor_meta_diaria_poa", functions.round("valor_meta_diaria_poa", 2))

Criando dataframe unindo as duas metas:

1df_metas_final = df_metas_pev.join(df_metas_poa, ["mes", "negocio"])

Fazendo join entre a base de datas e a de metas:

1df_datas_ajuste = spark.sql("select * from indicadores_mercado.tb_datas") 2df_metas_ajustado = df_datas_ajuste.join(df_metas_final, ["mes", "negocio"], "left")

Por fim, criando a tabela final. Fazendo query na database para vendas e join com o dataframe de metas:

1df_vendas_metas_deca = df_vendas.join(df_metas_ajustado, ["data", "mes", "negocio"])

Criando particionamento, para posteriormente calcular as metas acumuladas por mês e negócio:

1part = (Window.partitionBy("negocio", "mes").orderBy("data").rangeBetween(Window.unboundedPreceding, 0))

Criando as metas acumuladas utilizando o particionamento:

1df_vendas_metas_deca_acumuladas = df_vendas_metas_deca.withColumn("metas_acumuladas_pev", functions.sum("valor_meta_diaria_pev").over(part)).withColumn("metas_acumuladas_poa", functions.sum("valor_meta_diaria_poa").over(part))

Sobrescrevendo a tabela final na database indicadores_mercado:

1df_vendas_metas_deca_acumuladas.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("indicadores_mercado.tb_indicadores_vendas_deca")

19.3 Base Final:

col_name

data_type

data

date

mes

date

negocio

string

receita_liquida_vendas

double

receita_liquida_vendas_acumuladas

double

valor_meta_diaria_pev

double

valor_meta_diaria_poa

double

metas_acumuladas_pev

double

metas_acumuladas_poa

double