Guideline - Devs (Databricks)

Orientação para criação de branches no Bitbucket

Referencia de como integrar o Bitbucket no Databricks: Integração Bitbucket com Databricks

Todas as nossas branches deverão ser de um dos quatro tipos abaixo.

Release (implantação de um novo processo/integração)
Feature (novos recursos/integrações de um processo que já existe)
Hotfix (correção/ajuste de processos que não apresentam erro. Melhoria ou ajuste de uma regra, por exemplo)
Bugfix (correção de erro(s) em produção. Processos impactados e prejudicados por alguma exceção)

Na sequência deveremos informar o número da tarefa no Jira e o nome do responsável.

Referencia de como integrar o Bitbucket com o Jira: Integração Bitbucket com JIRA

Exemplo:

release/DGA-861/lucas_falotico

Uma vez que essa branch teve seu desenvolvimento concluído e refletido nas branchs develop e master, poderemos excluir do bitbucket.

Check List de Deploy para Develop e Produção

Solicitações para arquitetura:
- Criação de PR
  - Informar o link da PR
- Criação de Secret
  - informar ESCOPO e CHAVE (scope / key). Se tiverem dúvidas quanto a isso, perguntem.
- Liberação de JOB e Mount
  - Nome do JOB
  - Nome do Mount Utilizado
- Execução de Job em Produção
  - Informar para executar o JOB após a execução da PR (GARANTIR que o processo foi executado com sucesso em DEV)
- Carga de Tabelas (Entre Ambientes)
  - Informar todas as tabelas necessárias, sejam tabelas finais ou estáticas do processo.
  - Informar ambiente origem e ambiente destino
  - Informar se têm ou não partição
  - Garantir que tenham o Create Table If Not Exists nos notebooks de processamento
Validação em DEV dos processos desenvolvidos antes da PR para MASTER
Incluir Evidências de Execução do JOB Databricks na Tarefa Jira (ou na própria Pull Request). Essa evidência facilitará na validação da alteração pelo APROVADOR da PR.

Como acessar um banco de dados MySql

from pyspark.sql.functions import *

host = ""
user = ""
password = ""
port = ""
dbDatabase = ""
driver= "com.mysql.jdbc.Driver"
url= "jdbc:mysql://"+host+":"+port+"/"+dbDatabase+""
 
def read_table_mysql(sqlQuery):
 
    try:
        df = spark.read.format("jdbc")\
              .option("driver", driver)\
              .option("url", url)\
              .option("query", sqlQuery)\
              .option("user", user)\
              .option("password", password)\
              .load()

        return df
    
    except Exception as ex:
        print(f"❌ Erro na leitura da query")
        print(ex)
        pass

Como acessar um banco de dados Postgres

from pyspark.sql.functions import *

host = ""
dbname = ""
user = ""
password = ""
port = 5432

bucket_name = dbutils.secrets.get(scope = 'PROCESS_ETL_ANALYTICKS_DATABRICKS_API', key = 'BUCKET_TEMP_DATA_TRANSFER')

aws_access = dbutils.secrets.get(scope = 'AWS_SVC_S3_DATA_TRANSFER', key = 'ACCESS_KEY')
aws_secret = dbutils.secrets.get(scope = 'AWS_SVC_S3_DATA_TRANSFER', key = 'SECRET_KEY')
sc._jsc.hadoopConfiguration().set("fs.s3a.access.key", aws_access)
sc._jsc.hadoopConfiguration().set("fs.s3a.secret.key", aws_secret)

conn_string = f'jdbc:postgresql://{host}:{port}/{dbname}?user={user}&password={password}'

def query_postgres(select, nome_tabela):
    
    try:

        tempdir = f"s3a://{bucket_name}/Redshift/PRODUTOS-DIGITAIS/{tabela}"

        df = (spark.read.format("jdbc") 
                .option("url", conn_string) 
                .option("forward_spark_s3_credentials", "true") 
                .option("query", select) 
                .option("tempdir", tempdir) 
                .option("driver", "org.postgresql.Driver")
                .load())
        return df

    except Exception as ex:
        print(f"❌ Erro na leitura da query")
        print(ex)
        pass

Como acessar um banco de dados Sql Server

# DBTITLE 1,Configuração SQL Server
host = dbutils.secrets.get(scope = 'DB_ETIQUETA_SQL_SERVER', key = 'HOST')
database = dbutils.secrets.get(scope = 'DB_ETIQUETA_SQL_SERVER', key = 'DATABASE')
password = dbutils.secrets.get(scope = 'DB_ETIQUETA_SQL_SERVER', key = 'PASS')
user = dbutils.secrets.get(scope = 'DB_ETIQUETA_SQL_SERVER', key = 'USER')

server_name = f"jdbc:sqlserver://{host}:1433"
url = server_name + ";" + "databaseName=" + database + ";encrypt=true;trustServerCertificate=true;"

TABLE_TO_WRITE = '<NOME DA TABELA>'

#READ TABLE
remote_table = (spark.read
  .format("jdbc")
  .option("url", url)
  .option("dbtable", TABLE_TO_WRITE)
  .option("user", user)
  .option("password", password)
  .load()
)

# WRITE TABLE
df.distinct().write.format("jdbc")\
    .mode("overwrite")\
    .option("truncate","true")\
    .option("url", url)\
    .option("dbtable", TABLE_TO_WRITE)\
    .option("user", user)\
    .option("password", password) \
    .option("schemaCheckEnabled", "false")\
    .save()