Conjunto de notebooks que têm 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.

1. Indicadores Carteira Deca

Indicadores para pedidos em 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:

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

Renomeando as colunas:

colunas_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"]
df_pedidos_carteira = pedidos_carteira.toDF(*colunas_pedidos_carteira)

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

df_carteira = spark.sql("""
    select
        to_date(data_atualizacao) as data_referencia,
        (case when codigo_setor_atividade == 'HY' then 'hydra'
            when codigo_setor_atividade == 'MS' then 'metais'
            when codigo_setor_atividade == 'CS' then 'loucas' end) as negocio,
        sum(case when status_carteira in('Bloqueio Adm.', 'Credito', 'Limbo', 'Limbo Programado', 'Não classificado') then valor_pendente else 0 end) as valor_bloqueado,
        sum(case when status_carteira in('Programado', 'Remetido') then valor_pendente else 0 end) as valor_livre
   
    from pedidos_carteira
     
    where to_date(data_atualizacao) == current_date()
        and codigo_setor_atividade in('CS', 'HY', 'MS')

    group by 1, 2
""")

Fazendo append na tabela final na database indicadores_mercado:

df_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.

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:

df_devolucoes_deca = spark.sql("""
    select distinct
        to_date(date_format(data_competencia, 'yyyy-MM-01')) as mes,
        (case when codigo_setor_atividade == 'MS' then 'metais'
             when codigo_setor_atividade == 'CS' then 'loucas'
             when codigo_setor_atividade == 'HY' then 'hydra'
             when codigo_setor_atividade == '01' then 'revestimento' end) as negocio,
        sum(case when tipo_documento_venda == 'S2' then valor_receita_liquida else 0 end) as valor_estorno,
        sum(case when status_ordem_venda == 'DEVOLUÇÃO' then abs(valor_receita_liquida) else 0 end) as valor_devolucao,
        sum(case when status_ordem_venda == 'FATURAMENTO' then valor_receita_liquida else 0 end) as valor_faturamento
        
    from large.tb_resultado_comercial
    
    where data_competencia between '2019-01-01' and current_date() - 1
        and codigo_setor_atividade in('MS', 'CS', 'HY', '01')
    
    group by 1, 2
    
    order by 1, 2 asc
""")

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

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

Sobrescrevendo a tabela final na database indicadores_mercado:

df_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.

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:

df_devolucao_madeira = spark.sql("""
    select
        perio as mes,
        (case when prctr like '%MDP' then 'mdp'
            when prctr like '%MDF' then 'mdf' else 'paineis' end) as negocio,
        sum(case when fkart in('ZREB', 'ZROB') then abs((vv089) - (vv001 + vv002 + vv003 + vv004)) else 0 end) as valor_devolucao,
        sum(case when fkart not in('ZREB', 'ZROB') then (vv089) - (vv001 + vv002 + vv003 + vv004) else 0 end) as valor_faturado
        
    from analytics_prd.custos_rem
    
    where spart == 'CH'
        and (prctr like '%MDP' or prctr like '%MDF')
    
    and perio >= 2019001
    
    group by 1, 2
    
    order by 1, 2
""")

Sobrescrevendo a tabela final na database indicadores_mercado:

df_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.

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:

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

Fazendo o download dos arquivos:

download_arquivo("/sites/RelatriosControladoria/Documentos%20Compartilhados/Report%20Book%20Diretoria/Fechamento%20Gerencial%202022.xlsm", "Fechamento_Gerencial_2022.xlsm")
download_arquivo("/sites/RelatriosControladoria/Documentos%20Compartilhados/Report%20Book%20Diretoria/Fluxo%20de%20Caixa%20Oficial%202022.xlsx", "Fluxo_de_Caixa_Oficial_2022.xlsx")
download_arquivo("/sites/RelatriosControladoria/Documentos%20Compartilhados/Report%20Book%20Diretoria/Forecast%20Consolidador.xlsx", "Forecast_3_9_2022.xlsx")
download_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:

#parâmetros da função: range de células e negócio("madeira", "deca", "revestimento" ou "consolidado")
def ler_dre(celulas, negocio):
    #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)
    #transpondo a tabela utilizando .transpose()
    #selecionando colunas utilizando .iloc[]
    #resetando index
    df = pd.read_excel(fechamento_gerencial, "DRE CONSOL", skiprows = 1, nrows = 146, usecols = celulas, index_col = 0).transpose().iloc[:, [6, 127]].reset_index()
    #renomeando colunas
    df.columns = dre_header
    #transformando em dataframe do spark
    df = spark.createDataFrame(df)
    #criando coluna "negocio"
    df = df.selectExpr(dre_header[0], "'{}' as negocio".format(negocio), dre_header[1], dre_header[2])
    #tratando coluna "mes"
    df = df.withColumn("mes", when(df.mes.like("JAN%"), "01/01/" + ano_atual)
                           .when(df.mes.like("FEV%"), "01/02/" + ano_atual)
                           .when(df.mes.like("MAR%"), "01/03/" + ano_atual)
                           .when(df.mes.like("ABR%"), "01/04/" + ano_atual)
                           .when(df.mes.like("MAI%"), "01/05/" + ano_atual)
                           .when(df.mes.like("JUN%"), "01/06/" + ano_atual)
                           .when(df.mes.like("JUL%"), "01/07/" + ano_atual)
                           .when(df.mes.like("AGO%"), "01/08/" + ano_atual)
                           .when(df.mes.like("SET%"), "01/09/" + ano_atual)
                           .when(df.mes.like("OUT%"), "01/10/" + ano_atual)
                           .when(df.mes.like("NOV%"), "01/11/" + ano_atual)
                           .when(df.mes.like("DEZ%"), "01/12/" + ano_atual)
                           .otherwise(None))
    df = df.withColumn("mes", to_date("mes", "dd/MM/yyyy"))
    return df

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

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

Fazendo união de todos os dataframes gerados:

df_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:

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

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

df_eva_madeira = ler_eva("GG:GT", "madeira")
df_eva_deca = ler_eva("BA:BN", "deca")
df_eva_revestimento = ler_eva("EY:FL", "revestimento")
df_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:

df_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:

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

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

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

Fazendo união de todos os dataframes gerados:

df_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:

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

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

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

Fazendo união de todos os dataframes gerados:

df_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:

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

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

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

Fazendo união de todos os dataframes gerados:

df_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:

def ler_forecast_eva(negocio):
    #desinindo constantantes para cada negócio para utilizar posteriormente o iloc[] e localizar os dados necessários
    #esses valores pra neg são como os negócios são identificados no arquivo
    if negocio == "madeira":
        neg = "Mad. Total"
    elif negocio == "deca":
        neg = "Deca"
    elif negocio == "revestimento":
        neg = "RC"
    elif negocio == "consolidado":
        neg = "Consol"
    df = pd.read_excel(forecast, "Base EVA ROIC", skiprows = 4, nrows = 130, usecols = "B:Q")
    #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"
    df = df.where(
        (df.iloc[:, 0] == neg) &
        (df.iloc[:, 1] == "Mês") &
        (df.iloc[:, 2] == "RECORRENTE") &
        (df.iloc[:, 3] == "EVA")
    ).dropna()
    df = df.transpose().iloc[4: ].reset_index()
    df.columns = frc_eva_header
    df = spark.createDataFrame(df)
    df = df.selectExpr(frc_eva_header[0], "'{}' as negocio".format(negocio), frc_eva_header[1])
    df = df.withColumn("mes", when(df.mes == 1, "01/01/" + ano_atual)
                           .when(df.mes == 2, "01/02/" + ano_atual)
                           .when(df.mes == 3, "01/03/" + ano_atual)
                           .when(df.mes == 4, "01/04/" + ano_atual)
                           .when(df.mes == 5, "01/05/" + ano_atual)
                           .when(df.mes == 6, "01/06/" + ano_atual)
                           .when(df.mes == 7, "01/07/" + ano_atual)
                           .when(df.mes == 8, "01/08/" + ano_atual)
                           .when(df.mes == 9, "01/09/" + ano_atual)
                           .when(df.mes == 10, "01/10/" + ano_atual)
                           .when(df.mes == 11, "01/11/" + ano_atual)
                           .when(df.mes == 12, "01/12/" + ano_atual)
                           .otherwise(None))
    df = df.withColumn("mes", to_date("mes", "dd/MM/yyyy"))
    return df

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

df_frc_eva_madeira = ler_forecast_eva("madeira")
df_frc_eva_deca = ler_forecast_eva("deca")
df_frc_eva_revestimento = ler_forecast_eva("revestimento")
df_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:

df_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:

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

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

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

Fazendo união de todos os dataframes gerados:

df_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:

df_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:

df_poa = df_poa.select(
    "mes",
    "negocio",
    col("poa_receita_liquida_vendas").cast(DoubleType()),
    col("poa_ebitda_recorrente").cast(DoubleType()),
    col("poa_eva_recorrente").cast(DoubleType()),
    col("poa_fluxo_caixa_livre_total").cast(DoubleType())
).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:

campos_join = ["mes", "negocio"]
df_indicadores_financeiros = df_dre.join(
    df_eva, campos_join
).join(
    df_fc, campos_join
).join(
    df_pmp, campos_join
).join(
    df_frc, campos_join
).join(
    df_frc_eva, campos_join
).join(
    df_frc_fc, campos_join
).join(
    df_poa, campos_join
).orderBy("negocio", "mes")

Sobrescrevendo a tabela final no database :

df_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.

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:

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

Lendo o arquivo:

df_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:

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

Sobrescrevendo a tabela final na database indicadores_mercado:

df_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.

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:

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

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

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

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

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

Fazendo o download do arquivo .tdsx:

with servidor.auth.sign_in(autenticador):
    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:

with zipfile.ZipFile(diretorio_download + nome_database + ".tdsx", "r") as arquivo_zipado:
    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:

with HyperProcess(telemetry = Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU) as hyper:
    with Connection(endpoint = hyper.endpoint, database = diretorio_download + "Data/Extracts/" + nome_database + ".hyper") as conexao:
        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:
            resultado_consulta = list(resultado)

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

esquema_margem = StructType([
    StructField('mes', StringType(), True),
    StructField('produto', StringType(), True),
    StructField('mercado', StringType(), True),
    StructField('receita', DoubleType(), True),
    StructField('quantidade', DoubleType(), True),
    StructField('valor_cit', DoubleType(), True),
    StructField('valor_cco', DoubleType(), True)
])

Criando o dataframe:

df_margem = spark.createDataFrame(resultado_consulta, esquema_margem)

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

df_margem_calculos = spark.sql("""
    select
        mes,
        (case when produto like 'MDF%' then 'mdf' when produto like 'MDP%' then 'mdp' else 'outros' end) as produto,
        (sum(receita) / sum(quantidade)) as receita_m3,
        (sum(valor_cit) / sum(quantidade)) as cit_m3,
        (sum(valor_cco) / sum(quantidade)) as cco_m3
       
    from margem_view
    
    group by 1, 2
    
    order by 1, 2 asc
""")

Aplicando mais algumas regras de negócio:

df_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:

df_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

Dados Internos Dexco

1. Indicadores Carteira Deca

1.1 Origem:

1.2 Transformação:

1.3 Base Final:

2. Indicadores Devolução Deca

2.1 Origem:

2.2 Transformação:

2.3 Base Final:

3. Indicadores Devolução Madeira

3.1 Origem:

3.2 Transformação:

3.3 Base Final:

4. Indicadores Financeiros

4.1 Origem:

4.2 Transformação:

4.2.1 Arquivo Fechamento_Gerencial_2022.xlsm

4.2.1.1 Indicadores do DRE

4.2.1.2 Indicadores do EVA

4.2.2 Arquivo Fluxo_de_Caixa_Oficial_2022.xlsx

4.2.3 Arquivo Forecast_3_9_2022.xlsx

4.2.4 Arquivo FCST_2_10___Fluxo_de_Caixa_Livre_2022.xlsx

4.2.5 Arquivo indicadores_poa_2022.csv

4.2.6 Base Final

4.3 Base Final:

5. Indicadores Gente

5.1 Origem:

5.2 Transformação:

5.3 Base Final:

7. Indicadores Margem Madeira

7.1 Origem:

7.2 Transformação:

7.3 Base Final: