2 _ [LAB - Hotelaria] Integração do Neon PostgreSQL com Databricks Free Edition

Após definir a organização dos ambientes e camadas no Databricks, o próximo passo do laboratório será criar uma fonte de dados relacional real para simular um cenário próximo de produção.

Neste projeto, usaremos o Neon PostgreSQL como banco transacional de origem e o Databricks Free Edition como plataforma Lakehouse para ingestão, transformação e organização dos dados nas camadas landing, bronze, silver e gold.

A ideia é simular o fluxo:

Neon PostgreSQL
      ↓
Databricks Connector / JDBC / Lakehouse Federation
      ↓
hotelaria_dev.landing
      ↓
hotelaria_dev.bronze
      ↓
hotelaria_dev.silver
      ↓
hotelaria_dev.gold

Por que usar o Neon PostgreSQL neste laboratório?

O objetivo é evitar um laboratório “artificial”, onde os dados já nascem dentro do próprio Databricks.

Em projetos reais, os dados normalmente vêm de sistemas transacionais, ERPs, CRMs, bancos relacionais, APIs ou arquivos. Por isso, usar um PostgreSQL externo ajuda a simular melhor um processo real de engenharia de dados.

O Neon é uma boa opção para laboratório porque oferece PostgreSQL gerenciado em nuvem, acesso por connection string, porta padrão 5432 e conexão com SSL/TLS obrigatório. A própria documentação do Neon orienta obter a string pelo botão Connect no dashboard do projeto, selecionando branch, role e database. (Neon)

Papel do Neon dentro da arquitetura

Neste projeto, o Neon será tratado como o source system.

Source System:
  Neon PostgreSQL

Database:
  hotelaria

Schema:
  public

Tabelas:
  hoteis
  quartos
  hospedes
  reservas
  consumos
  faturas
  reservas_ota

Essas tabelas representam entidades operacionais de um cenário de hotelaria, como hotéis, quartos, hóspedes, reservas, consumos, faturas e reservas vindas de OTAs. Elas serão usadas como base para construir o fluxo medallion no Databricks.

Modelo no Databricks

Como o Databricks Free Edition possui apenas recursos serverless e limitações específicas de uso, vamos manter a separação de ambientes por catálogo dentro de um único workspace. A documentação da Databricks informa que a Free Edition usa compute serverless, não suporta configurações customizadas de compute e restringe acesso externo à internet a domínios confiáveis. (Documentação Databricks)

A estrutura do laboratório será:

hotelaria_dev
├── landing
├── bronze
├── silver
└── gold

hotelaria_prod
├── landing
├── bronze
├── silver
└── gold

Para esta etapa, vamos trabalhar primeiro com:

hotelaria_dev.landing
hotelaria_dev.bronze

Depois evoluímos para:

hotelaria_dev.silver
hotelaria_dev.gold

Passo 1 — Criar conta no Neon

Acesse o site do Neon e crie uma conta.

Você pode usar login com GitHub, Google ou e-mail.

Depois de entrar no painel:

1. Clique em New Project
2. Escolha um nome para o projeto
3. Escolha a região
4. Crie o projeto

Sugestão de nome:

hotelaria-etl-lab

Passo 2 — Criar ou usar o banco existente

Ao criar um projeto no Neon, normalmente já é criado um banco padrão, como:

neondb

Você pode usar esse banco ou criar um novo chamado:

hotelaria

Para criar o banco manualmente no SQL Editor do Neon:

create database hotelaria;

Depois, conecte-se ao banco hotelaria.

Se preferir simplificar o laboratório, pode usar o banco padrão neondb e criar as tabelas diretamente no schema public.

Minha recomendação para o laboratório:

Projeto Neon: hotelaria-etl-lab
Database: hotelaria
Schema: public

Passo 3 — Criar usuário específico para ingestão

Em projeto real, não é uma boa prática usar o usuário administrador da base para integração de dados.

Crie um usuário específico para leitura:

create user databricks_ingest with password 'troque_essa_senha_forte';

grant connect on database hotelaria to databricks_ingest;

grant usage on schema public to databricks_ingest;

grant select on all tables in schema public to databricks_ingest;

alter default privileges in schema public
grant select on tables to databricks_ingest;

A Databricks também recomenda criar um usuário separado para ingestão/replicação quando conecta ao PostgreSQL, mantendo credenciais governadas e seguras. (Microsoft Learn)

Para o laboratório, se você estiver apenas testando, pode começar com o usuário padrão do Neon e depois refatorar para o usuário databricks_ingest.

Passo 4 — Criar as tabelas no Neon PostgreSQL

Execute o script abaixo no SQL Editor do Neon.

drop table if exists public.reservas_ota cascade;
drop table if exists public.consumos cascade;
drop table if exists public.faturas cascade;
drop table if exists public.reservas cascade;
drop table if exists public.quartos cascade;
drop table if exists public.hospedes cascade;
drop table if exists public.hoteis cascade;

create table public.hoteis (
    hotel_id integer primary key,
    nome_hotel text,
    endereco text,
    cidade text,
    estado text,
    estrelas integer,
    numero_quartos integer,
    comodidades text,
    telefone text,
    email_contato text,
    data_abertura timestamp,
    horario_checkin text,
    horario_checkout text,
    categoria_hotel text,
    tipo_hotel text,
    ano_fundacao integer,
    capacidade_total integer,
    possui_acessibilidade boolean,
    certificacoes text,
    latitude double precision,
    longitude double precision,
    descricao_hotel text,
    numero_funcionarios integer,
    insert_date timestamp default current_timestamp,
    update_date timestamp default current_timestamp
);

create table public.hospedes (
    hospede_id integer primary key,
    nome_completo text,
    cpf text,
    data_nascimento timestamp,
    email text,
    telefone text,
    estado text,
    nacionalidade text,
    data_cadastro timestamp,
    programa_fidelidade text,
    profissao text,
    tipo_documento text,
    numero_documento text,
    empresa text,
    eh_viajante_frequente boolean,
    preferencias_hospede text,
    restricoes_alimentares text,
    data_ultima_hospedagem timestamp,
    total_hospedagens integer,
    insert_date timestamp default current_timestamp,
    update_date timestamp default current_timestamp
);

create table public.quartos (
    quarto_id integer primary key,
    hotel_id integer references public.hoteis(hotel_id),
    numero_quarto text,
    tipo_quarto text,
    capacidade_maxima integer,
    preco_diaria_base double precision,
    andar integer,
    vista text,
    comodidades_quarto text,
    possui_ar_condicionado boolean,
    tamanho_quarto text,
    status_manutencao text,
    ultima_manutencao timestamp,
    eh_smoke_free boolean,
    possui_kit_boas_vindas boolean,
    numero_camas integer,
    insert_date timestamp default current_timestamp,
    update_date timestamp default current_timestamp
);

create table public.reservas (
    reserva_id integer primary key,
    hospede_id integer references public.hospedes(hospede_id),
    quarto_id integer references public.quartos(quarto_id),
    hotel_id integer references public.hoteis(hotel_id),
    data_reserva timestamp,
    data_checkin timestamp,
    data_checkout timestamp,
    numero_noites integer,
    numero_adultos integer,
    numero_criancas integer,
    canal_reserva text,
    status_reserva text,
    data_cancelamento timestamp,
    solicitacoes_especiais text,
    valor_total_estadia double precision,
    motivo_viagem text,
    motivo_cancelamento text,
    taxa_limpeza double precision,
    taxa_turismo double precision,
    avaliacao_hospede double precision,
    comentarios_hospede text,
    insert_date timestamp default current_timestamp,
    update_date timestamp default current_timestamp
);

create table public.consumos (
    consumo_id integer primary key,
    reserva_id integer references public.reservas(reserva_id),
    hospede_id integer references public.hospedes(hospede_id),
    hotel_id integer references public.hoteis(hotel_id),
    nome_servico text,
    data_consumo timestamp,
    quantidade integer,
    valor_total_consumo double precision,
    hora_consumo text,
    local_consumo text,
    funcionario_responsavel text,
    insert_date timestamp default current_timestamp,
    update_date timestamp default current_timestamp
);

create table public.faturas (
    fatura_id integer primary key,
    reserva_id integer references public.reservas(reserva_id),
    hospede_id integer references public.hospedes(hospede_id),
    data_emissao timestamp,
    data_vencimento timestamp,
    status_pagamento text,
    forma_pagamento text,
    subtotal_estadia double precision,
    subtotal_consumos double precision,
    descontos double precision,
    impostos double precision,
    valor_total double precision,
    data_pagamento timestamp,
    taxa_limpeza double precision,
    taxa_turismo double precision,
    taxa_servico double precision,
    numero_transacao text,
    insert_date timestamp default current_timestamp,
    update_date timestamp default current_timestamp
);

create table public.reservas_ota (
    ota_reserva_id integer primary key,
    reserva_id integer references public.reservas(reserva_id),
    ota_codigo_confirmacao text,
    ota_nome_convidado text,
    total_pago_ota double precision,
    taxa_comissao double precision,
    valor_liquido_recebido double precision,
    ota_solicitacoes_especificas text,
    insert_date timestamp default current_timestamp,
    update_date timestamp default current_timestamp
);

Passo 5 — Popular as tabelas com dados fictícios

Execute este script no Neon:

insert into public.hoteis (
    hotel_id, nome_hotel, endereco, cidade, estado, estrelas,
    numero_quartos, comodidades, telefone, email_contato,
    data_abertura, horario_checkin, horario_checkout,
    categoria_hotel, tipo_hotel, ano_fundacao, capacidade_total,
    possui_acessibilidade, certificacoes, latitude, longitude,
    descricao_hotel, numero_funcionarios
)
values
(1, 'Hotel Atlântico', 'Av. Beira Mar, 100', 'Salvador', 'BA', 4, 120, 'Piscina, Wi-Fi, Restaurante', '(71) 3000-1000', 'contato@hotelatlantico.com', '2010-01-15 00:00:00', '14:00', '12:00', 'Executivo', 'Urbano', 2010, 300, true, 'ISO 9001', -12.9714, -38.5014, 'Hotel voltado para turismo e negócios.', 85),
(2, 'Hotel Chapada Premium', 'Rua das Palmeiras, 45', 'Lençóis', 'BA', 5, 80, 'Spa, Trilhas, Restaurante', '(75) 3000-2000', 'contato@chapadapremium.com', '2015-06-20 00:00:00', '15:00', '11:00', 'Luxo', 'Resort', 2015, 220, true, 'Selo Turismo Sustentável', -12.5611, -41.3890, 'Hotel de luxo próximo à Chapada Diamantina.', 65);

insert into public.hospedes (
    hospede_id, nome_completo, cpf, data_nascimento, email,
    telefone, estado, nacionalidade, data_cadastro,
    programa_fidelidade, profissao, tipo_documento, numero_documento,
    empresa, eh_viajante_frequente, preferencias_hospede,
    restricoes_alimentares, data_ultima_hospedagem, total_hospedagens
)
values
(1, 'Ana Souza', '12345678901', '1990-05-10 00:00:00', 'ana.souza@email.com', '(71) 99999-0001', 'BA', 'Brasileira', '2024-01-10 10:00:00', 'Gold', 'Analista de Sistemas', 'RG', '1234567', 'Tech Bahia', true, 'Quarto alto e silencioso', 'Sem lactose', '2026-04-10 00:00:00', 6),
(2, 'Carlos Lima', '98765432100', '1985-09-22 00:00:00', 'carlos.lima@email.com', '(11) 98888-0002', 'SP', 'Brasileira', '2024-03-18 11:30:00', 'Silver', 'Consultor', 'CNH', '7654321', 'Lima Consultoria', false, 'Cama king size', null, '2026-03-05 00:00:00', 2),
(3, 'Mariana Costa', '45678912300', '1995-12-01 00:00:00', 'mariana.costa@email.com', '(31) 97777-0003', 'MG', 'Brasileira', '2025-02-01 09:00:00', 'Bronze', 'Designer', 'RG', '9988776', null, true, 'Vista para o mar', 'Vegetariana', null, 0);

insert into public.quartos (
    quarto_id, hotel_id, numero_quarto, tipo_quarto,
    capacidade_maxima, preco_diaria_base, andar, vista,
    comodidades_quarto, possui_ar_condicionado, tamanho_quarto,
    status_manutencao, ultima_manutencao, eh_smoke_free,
    possui_kit_boas_vindas, numero_camas
)
values
(1, 1, '0101', 'Standard', 2, 280.00, 1, 'Cidade', 'TV, Wi-Fi, Frigobar', true, '25m²', 'OK', '2026-01-10 00:00:00', true, true, 1),
(2, 1, '0802', 'Luxo', 3, 520.00, 8, 'Mar', 'TV, Wi-Fi, Frigobar, Banheira', true, '40m²', 'OK', '2026-02-15 00:00:00', true, true, 2),
(3, 2, '0205', 'Chalé Premium', 4, 750.00, 2, 'Montanha', 'Lareira, Wi-Fi, Varanda', true, '55m²', 'OK', '2026-01-25 00:00:00', true, true, 2);

insert into public.reservas (
    reserva_id, hospede_id, quarto_id, hotel_id, data_reserva,
    data_checkin, data_checkout, numero_noites, numero_adultos,
    numero_criancas, canal_reserva, status_reserva,
    data_cancelamento, solicitacoes_especiais, valor_total_estadia,
    motivo_viagem, motivo_cancelamento, taxa_limpeza, taxa_turismo,
    avaliacao_hospede, comentarios_hospede
)
values
(1, 1, 2, 1, '2026-04-01 10:00:00', '2026-04-10 14:00:00', '2026-04-13 12:00:00', 3, 2, 0, 'Site', 'Finalizada', null, 'Check-in antecipado se possível', 1560.00, 'Lazer', null, 80.00, 30.00, 4.8, 'Excelente estadia.'),
(2, 2, 1, 1, '2026-04-05 09:30:00', '2026-04-20 14:00:00', '2026-04-22 12:00:00', 2, 1, 0, 'Telefone', 'Confirmada', null, null, 560.00, 'Negócios', null, 50.00, 20.00, null, null),
(3, 3, 3, 2, '2026-04-08 16:00:00', '2026-05-01 15:00:00', '2026-05-04 11:00:00', 3, 2, 1, 'OTA', 'Cancelada', '2026-04-20 08:00:00', 'Berço infantil', 2250.00, 'Lazer', 'Mudança de planos', 100.00, 45.00, null, null);

insert into public.consumos (
    consumo_id, reserva_id, hospede_id, hotel_id, nome_servico,
    data_consumo, quantidade, valor_total_consumo, hora_consumo,
    local_consumo, funcionario_responsavel
)
values
(1, 1, 1, 1, 'Jantar Restaurante', '2026-04-10 20:30:00', 2, 180.00, '20:30', 'Restaurante', 'João Pereira'),
(2, 1, 1, 1, 'Serviço de Quarto', '2026-04-11 22:10:00', 1, 95.00, '22:10', 'Quarto 0802', 'Maria Oliveira'),
(3, 2, 2, 1, 'Lavanderia', '2026-04-21 09:00:00', 3, 60.00, '09:00', 'Lavanderia', 'Carlos Santos');

insert into public.faturas (
    fatura_id, reserva_id, hospede_id, data_emissao,
    data_vencimento, status_pagamento, forma_pagamento,
    subtotal_estadia, subtotal_consumos, descontos, impostos,
    valor_total, data_pagamento, taxa_limpeza, taxa_turismo,
    taxa_servico, numero_transacao
)
values
(1, 1, 1, '2026-04-13 12:10:00', '2026-04-13 23:59:59', 'Pago', 'Cartão de Crédito', 1560.00, 275.00, 50.00, 120.00, 1935.00, '2026-04-13 12:15:00', 80.00, 30.00, 20.00, 'TXN-10001'),
(2, 2, 2, '2026-04-22 12:05:00', '2026-04-22 23:59:59', 'Pendente', 'Pix', 560.00, 60.00, 0.00, 40.00, 730.00, null, 50.00, 20.00, 0.00, 'TXN-10002');

insert into public.reservas_ota (
    ota_reserva_id, reserva_id, ota_codigo_confirmacao,
    ota_nome_convidado, total_pago_ota, taxa_comissao,
    valor_liquido_recebido, ota_solicitacoes_especificas
)
values
(1, 3, 'OTA-ABC-20260501', 'Mariana Costa', 2250.00, 337.50, 1912.50, 'Solicitou berço infantil e quarto silencioso');

Passo 6 — Validar os dados no Neon

Execute:

select count(*) as total_hoteis from public.hoteis;
select count(*) as total_hospedes from public.hospedes;
select count(*) as total_quartos from public.quartos;
select count(*) as total_reservas from public.reservas;
select count(*) as total_consumos from public.consumos;
select count(*) as total_faturas from public.faturas;
select count(*) as total_reservas_ota from public.reservas_ota;

Também pode validar o relacionamento principal:

select
    r.reserva_id,
    h.nome_completo,
    ho.nome_hotel,
    q.numero_quarto,
    r.status_reserva,
    r.valor_total_estadia
from public.reservas r
left join public.hospedes h
    on r.hospede_id = h.hospede_id
left join public.hoteis ho
    on r.hotel_id = ho.hotel_id
left join public.quartos q
    on r.quarto_id = q.quarto_id;

Passo 7 — Obter a connection string no Neon

No painel do Neon:

1. Acesse o projeto hotelaria-etl-lab
2. Clique em Connect
3. Selecione a branch
4. Selecione o database hotelaria
5. Selecione o usuário
6. Copie a connection string

O Neon monta uma connection string semelhante a esta:

postgresql://usuario:senha@ep-xxxxxx.us-east-2.aws.neon.tech/hotelaria?sslmode=require

Para uso com Spark JDBC, vamos converter para este formato:

jdbc:postgresql://ep-xxxxxx.us-east-2.aws.neon.tech:5432/hotelaria?sslmode=require

O Neon usa a porta padrão do PostgreSQL, 5432, e exige SSL/TLS nas conexões. (Neon)

Passo 8 — Criar os catálogos e schemas no Databricks

No Databricks, execute:

create catalog if not exists hotelaria_dev;
create schema if not exists hotelaria_dev.landing;
create schema if not exists hotelaria_dev.bronze;
create schema if not exists hotelaria_dev.silver;
create schema if not exists hotelaria_dev.gold;

create catalog if not exists hotelaria_prod;
create schema if not exists hotelaria_prod.landing;
create schema if not exists hotelaria_prod.bronze;
create schema if not exists hotelaria_prod.silver;
create schema if not exists hotelaria_prod.gold;

Passo 9 — Opção A: criar conexão federada com PostgreSQL

A abordagem mais alinhada ao Unity Catalog é criar uma conexão com o PostgreSQL e depois um foreign catalog.

A Lakehouse Federation permite consultar dados de fontes externas sem migrar imediatamente todos os dados para o Lakehouse. No caso do PostgreSQL, a Databricks informa que é necessário criar uma conexão e um foreign catalog que espelha o banco externo no Unity Catalog. (Documentação Databricks)

Exemplo conceitual:

create connection if not exists conn_neon_hotelaria
type postgresql
options (
  host 'ep-proud-boat-aqsudtlr-pooler.c-8.us-east-1.aws.neon.tech',
  port '5432',
  user 'databricks_ingest',
  password 'SUA_NOVA_SENHA'
);

Depois:

create foreign catalog if not exists neon_hotelaria
using connection conn_neon_hotelaria
options (
  database 'hotelaria'
);

Depois teste:

show schemas in neon_hotelaria;

E depois:

show tables in neon_hotelaria.public;

Consulta de validação:

select *from neon_hotelaria.public.hoteis limit 10;

Ponto importante: dependendo dos recursos disponíveis no Free Edition, a Lakehouse Federation pode não estar disponível ou pode ter limitações. Se a conexão federada não funcionar, use a opção JDBC no notebook.

Passo 10 — Opção B: ler o Neon via PySpark JDBC

Essa é a abordagem mais direta para o laboratório.

Crie um notebook no Databricks:

notebooks/01_ingest_neon_postgres_to_landing.py

Script base:

jdbc_host = "ep-xxxxxx.us-east-2.aws.neon.tech"
jdbc_port = "5432"
jdbc_database = "hotelaria"

jdbc_url = (
    f"jdbc:postgresql://{jdbc_host}:{jdbc_port}/{jdbc_database}"
    "?sslmode=require"
)

jdbc_user = "databricks_ingest"
jdbc_password = "sua_senha"

connection_properties = {
    "user": jdbc_user,
    "password": jdbc_password,
    "driver": "org.postgresql.Driver"
}

tables = [
    "hoteis",
    "hospedes",
    "quartos",
    "reservas",
    "consumos",
    "faturas",
    "reservas_ota"
]

for table_name in tables:
    print(f"Lendo tabela origem: public.{table_name}")

    df = (
        spark.read
        .format("jdbc")
        .option("url", jdbc_url)
        .option("dbtable", f"public.{table_name}")
        .option("user", jdbc_user)
        .option("password", jdbc_password)
        .option("driver", "org.postgresql.Driver")
        .load()
    )

    print(f"Total de registros lidos em {table_name}: {df.count()}")

    target_table = f"hotelaria_dev.landing.{table_name}"

    (
        df.write
        .format("delta")
        .mode("overwrite")
        .option("overwriteSchema", "true")
        .saveAsTable(target_table)
    )

    print(f"Tabela gravada em: {target_table}")

O Spark possui suporte a leitura JDBC e mapeia tipos PostgreSQL para tipos Spark, como boolean para BooleanType, integer para IntegerType, bigint para LongType, double precision para DoubleType, numeric/decimal para DecimalType e timestamp para tipos de timestamp. (Apache Spark)

Passo 11 — Testar a conexão antes da ingestão completa

Antes de ingerir tudo, teste apenas uma tabela:

jdbc_host = "ep-xxxxxx.us-east-2.aws.neon.tech"
jdbc_port = "5432"
jdbc_database = "hotelaria"

jdbc_url = (
    f"jdbc:postgresql://{jdbc_host}:{jdbc_port}/{jdbc_database}"
    "?sslmode=require"
)

df = (
    spark.read
    .format("jdbc")
    .option("url", jdbc_url)
    .option("dbtable", "public.hoteis")
    .option("user", "databricks_ingest")
    .option("password", "sua_senha")
    .option("driver", "org.postgresql.Driver")
    .load()
)

display(df)

Depois valide:

df.printSchema()
print(df.count())

Se esse teste funcionar, avance para a ingestão completa.

Passo 12 — Adicionar metadados técnicos na Landing

Para deixar o laboratório mais próximo de um projeto real, podemos adicionar colunas de metadados no momento da ingestão:

from pyspark.sql.functions import current_timestamp, lit, sha2, concat_ws, col

def add_metadata(df, source_table):
    return (
        df
        .withColumn("_metadata_ingestion_at", current_timestamp())
        .withColumn("_metadata_source_system", lit("neon_postgresql"))
        .withColumn("_metadata_source_database", lit("hotelaria"))
        .withColumn("_metadata_source_schema", lit("public"))
        .withColumn("_metadata_source_table", lit(source_table))
    )

Script completo com metadados:

from pyspark.sql.functions import current_timestamp, lit

jdbc_host = "ep-xxxxxx.us-east-2.aws.neon.tech"
jdbc_port = "5432"
jdbc_database = "hotelaria"

jdbc_url = (
    f"jdbc:postgresql://{jdbc_host}:{jdbc_port}/{jdbc_database}"
    "?sslmode=require"
)

jdbc_user = "databricks_ingest"
jdbc_password = "sua_senha"

tables = [
    "hoteis",
    "hospedes",
    "quartos",
    "reservas",
    "consumos",
    "faturas",
    "reservas_ota"
]

for table_name in tables:
    print("=" * 80)
    print(f"Iniciando ingestão da tabela: public.{table_name}")

    df_source = (
        spark.read
        .format("jdbc")
        .option("url", jdbc_url)
        .option("dbtable", f"public.{table_name}")
        .option("user", jdbc_user)
        .option("password", jdbc_password)
        .option("driver", "org.postgresql.Driver")
        .load()
    )

    df_landing = (
        df_source
        .withColumn("_metadata_ingestion_at", current_timestamp())
        .withColumn("_metadata_source_system", lit("neon_postgresql"))
        .withColumn("_metadata_source_database", lit("hotelaria"))
        .withColumn("_metadata_source_schema", lit("public"))
        .withColumn("_metadata_source_table", lit(table_name))
    )

    target_table = f"hotelaria_dev.landing.{table_name}"

    (
        df_landing.write
        .format("delta")
        .mode("overwrite")
        .option("overwriteSchema", "true")
        .saveAsTable(target_table)
    )

    total_rows = spark.table(target_table).count()

    print(f"Tabela destino: {target_table}")
    print(f"Total de registros gravados: {total_rows}")
    print(f"Ingestão concluída para: {table_name}")

Passo 13 — Validar as tabelas no Databricks

Depois da ingestão:

select count(*) from hotelaria_dev.landing.hoteis;
select count(*) from hotelaria_dev.landing.hospedes;
select count(*) from hotelaria_dev.landing.quartos;
select count(*) from hotelaria_dev.landing.reservas;
select count(*) from hotelaria_dev.landing.consumos;
select count(*) from hotelaria_dev.landing.faturas;
select count(*) from hotelaria_dev.landing.reservas_ota;

Consulta de amostra:

select *
from hotelaria_dev.landing.reservas
limit 10;

Consulta com relacionamento:

select
    r.reserva_id,
    h.nome_completo,
    ho.nome_hotel,
    q.numero_quarto,
    r.status_reserva,
    r.valor_total_estadia
from hotelaria_dev.landing.reservas r
left join hotelaria_dev.landing.hospedes h
    on r.hospede_id = h.hospede_id
left join hotelaria_dev.landing.hoteis ho
    on r.hotel_id = ho.hotel_id
left join hotelaria_dev.landing.quartos q
    on r.quarto_id = q.quarto_id;

Passo 14 — Criar Bronze a partir da Landing

A Landing representa a captura inicial. A Bronze será a primeira camada Delta organizada para processamento.

Exemplo:

from pyspark.sql.functions import current_timestamp, lit

tables = [
    "hoteis",
    "hospedes",
    "quartos",
    "reservas",
    "consumos",
    "faturas",
    "reservas_ota"
]

for table_name in tables:
    source_table = f"hotelaria_dev.landing.{table_name}"
    target_table = f"hotelaria_dev.bronze.{table_name}"

    print(f"Processando {source_table} → {target_table}")

    df = spark.table(source_table)

    df_bronze = (
        df
        .withColumn("_metadata_bronze_loaded_at", current_timestamp())
        .withColumn("_metadata_layer", lit("bronze"))
    )

    (
        df_bronze.write
        .format("delta")
        .mode("overwrite")
        .option("overwriteSchema", "true")
        .saveAsTable(target_table)
    )

    print(f"Tabela criada: {target_table}")

Passo 15 — Organização esperada após a ingestão

Ao final dessa etapa, teremos:

Neon PostgreSQL
└── hotelaria.public
    ├── hoteis
    ├── hospedes
    ├── quartos
    ├── reservas
    ├── consumos
    ├── faturas
    └── reservas_ota

No Databricks:

hotelaria_dev
├── landing
│   ├── hoteis
│   ├── hospedes
│   ├── quartos
│   ├── reservas
│   ├── consumos
│   ├── faturas
│   └── reservas_ota
│
└── bronze
    ├── hoteis
    ├── hospedes
    ├── quartos
    ├── reservas
    ├── consumos
    ├── faturas
    └── reservas_ota

Observação importante sobre o Free Edition

Como a Free Edition restringe o acesso externo à internet a domínios confiáveis, a conexão direta com o host do Neon pode depender do suporte atual do ambiente serverless. (Documentação Databricks)

Se a conexão falhar por bloqueio de rede, o laboratório ainda pode seguir por uma dessas alternativas:

1. Exportar os dados do Neon para CSV e subir no Databricks
2. Usar arquivo Parquet/CSV como fallback temporário
3. Testar via Lakehouse Federation, se disponível no ambiente
4. Migrar depois para uma edição paga ou workspace corporativo

Mas o caminho ideal para o laboratório é tentar primeiro:

Neon PostgreSQL → Databricks JDBC/Connector → Landing Delta

Resumo da decisão arquitetural

Neste projeto:

Neon PostgreSQL = banco transacional de origem
Databricks Free Edition = ambiente Lakehouse de laboratório
hotelaria_dev = catálogo de desenvolvimento
landing = camada inicial de ingestão
bronze = camada bruta estruturada em Delta
silver = camada tratada e padronizada
gold = camada analítica para consumo

Fluxo esperado:

public.hoteis        → hotelaria_dev.landing.hoteis        → hotelaria_dev.bronze.hoteis
public.hospedes      → hotelaria_dev.landing.hospedes      → hotelaria_dev.bronze.hospedes
public.quartos       → hotelaria_dev.landing.quartos       → hotelaria_dev.bronze.quartos
public.reservas      → hotelaria_dev.landing.reservas      → hotelaria_dev.bronze.reservas
public.consumos      → hotelaria_dev.landing.consumos      → hotelaria_dev.bronze.consumos
public.faturas       → hotelaria_dev.landing.faturas       → hotelaria_dev.bronze.faturas
public.reservas_ota  → hotelaria_dev.landing.reservas_ota  → hotelaria_dev.bronze.reservas_ota

Esse passo torna o laboratório mais realista, porque simula o consumo de dados a partir de um banco operacional externo, mantendo a mesma lógica usada em projetos reais de engenharia de dados.