Notebook que tem como objetivo de fazer o upload e criar algumas bases em database do Databricks. Esse processo é temporário, pois visa suprir a urgência do projeto para iniciar testes com esses dados. Posteriormente, quando o time de engenharia tiver disponibilizado essas bases no Databricks, esse processo não será mais necessário e a criação das views será ajustada.

• Notebook: upload_temporario_dimensoes_sap

• Bases finais: margin_analysis_temp.T137T, margin_analysis_temp.TGSBT, margin_analysis_temp.TVV1T, margin_analysis_temp.TVV2T, margin_analysis_temp.TVV3T, margin_analysis_temp.T001, margin_analysis_temp.TINCT, margin_analysis_temp.TVKOT, margin_analysis_temp.TPART, margin_analysis_temp.TVV4T e margin_analysis_temp.TVV5T.

1. Origem

Todas as bases têm origem na transação SE16N do SAP (conexão PEQ, servidor 172.58.1.52, instância 00). As tabelas extraídas foram: T137T, TGSBT, TVV1T, TVV2T, TVV3T, T001, TINCT, TVKOT, TPART, TVV4T e TVV5T. Único filtro aplicado na extração (quando aplicável) foi o idioma = “PT”.

Foram feitas as extrações em CSV e feito o upload no Sharepoint, para posterior leitura no processo.

2. Leitura arquivos e criação das tabelas

Primeiramente, são criadas as duas databases que serão utilizadas no projeto. Na database margin_analysis ficarão as views para serem consumidas para o projeto e na database margin_analysis_temp ficarão as tabelas temporárias criadas nesse processo.

spark.sql("""create database if not exists margin_analysis""")
spark.sql("""create database if not exists margin_analysis_temp""")

Foi criada uma função para fazer o download de cada arquivo para diretório do dbfs e criação das respectivas tabelas finais na database margin_analysis_temp. O nome da tabela final é o mesmo nome do arquivo, que representa a mesma nomenclatura no SAP.

def download_arquivo(nome_arquivo):
    ctx = ClientContext(site_sharepoint).with_credentials(UserCredential(usuario, senha))
    web = ctx.load(ctx.web).execute_query()
    response = File.open_binary(ctx, "/sites/arquivos_projeto_margin/Documentos Partilhados/" + nome_arquivo + ".csv")
    response.raise_for_status()
    with open("/dbfs/FileStore/margin_analysis/" + nome_arquivo, "wb") as arquivo:
        arquivo.write(response.content)
    df = spark.read.format("csv").option("encoding", "ISO-8859-1").option("header", True).load("dbfs:/FileStore/margin_analysis/" + nome_arquivo + ".csv", sep = ";")
    df.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("margin_analysis_temp." + nome_arquivo)

Listando e processando os arquivos.

lista_arquivos = ["T137T", "TGSBT", "TVV1T", "TVV2T", "TVV3T", "T001", "TINCT", "TVKOT", "TPART", "TVV4T", "TVV5T"]
for arquivo in lista_arquivos:
    download_arquivo(arquivo)