Guide: DATTA Multi-Source Notebook

Preview — feature under development. Behavior, screens and contracts may change without notice between releases.

The interesting question almost never lives in a single source: the indicator is in SQL, the relationship is in the graph and the context is in the documents. The DATTA Notebook is the iterative analytical environment where all three coexist — you combine Trino, Neo4j, OpenSearch and PySpark in the same file, with the connections already available in the environment, and reach a reproducible result without switching tools at every step.

This guide shows which source answers which kind of question, the query patterns worth copying, and a complete integrated risk analysis example.

Which source to use for each question

Source	Language	Storage	Typical time	Best for
Trino 480	SQL	Parquet/ORC on S3/GCS	~1–10s	Massive aggregations, ad-hoc analyses
Neo4j	Cypher	Graph in memory/disk	~100–500ms	Relationships, lineage, impact
OpenSearch	Lucene	Inverted indexes	~100–500ms	Full text, search, facets
PySpark	Python	Memory/Parquet	~100–500ms	ML, transformations, correlations

1. Opening and importing a notebook

DATTA's notebook environment is served by JupyterHub.

Option A: through the interface (recommended)

Open http://<endereco-da-plataforma>/jupyter.
Click Upload and select the file docs/notebook-exemplo-multi-fonte.dattanb.
The notebook is loaded into your personal JupyterHub server and can be run cell by cell right away.

Option B: through the notebooks folder

If the notebooks volume is mounted on your workstation, just copy the file into it:

bash

cp docs/notebook-exemplo-multi-fonte.dattanb /mnt/notebooks/

If you are unsure about the mounted path in your installation, talk to the platform administrator.

2. What changed: from Impala to Trino

The notebook's SQL engine is Trino 480, which replaced Impala. If you have old notebooks, adjust the queries to the new table-name format (catalog.schema.table):

Before (Impala)

sql

-- Impala
SELECT * FROM `default`.`tabela` 
-- Engine: Teradata Impala 4.0.0
-- Storage: GCS gs://datta-object (Parquet only)

Now (Trino)

sql

-- Trino with the Hive catalog
SELECT * FROM hive.database.tabela
-- Engine: Trino 480
-- Storage: HDFS + S3/GCS (Parquet, ORC, Iceberg)

Available catalogs

sql

-- List catalogs
SHOW CATALOGS;

-- Trino catalogs:
-- 1. hive        → Legacy Hive tables in Parquet
-- 2. iceberg     → Apache Iceberg (ACID, time-travel)
-- 3. memory      → In-memory tables (temp/staging)

3. What exists in each source

sql

-- Schema: hive.assistencia_social
-- Parquet tables with ~23M records in total

SELECT * FROM hive.assistencia_social.cidadaos;
-- Columns: id, nome, cpf, nis, data_nascimento, sexo, renda, municipio, bairro, status
-- Records: ~2.3M

SELECT * FROM hive.assistencia_social.beneficios;
-- Columns: id, programa, valor, situacao, titular_cpf, titular_nome, banco, data_inicio, tipo
-- Records: ~5.1M

SELECT * FROM hive.assistencia_social.transacoes;
-- Columns: id, tipo, valor, data_transacao, programa, conta, municipio, status, cidadao_cpf
-- Records: ~12.4M

SELECT * FROM hive.assistencia_social.eventos;
-- Columns: id, tipo, data_evento, descricao, cidadao_cpf, resultado
-- Records: ~3.8M

SELECT * FROM hive.assistencia_social.alertas;
-- Columns: id, severidade, tipo, descricao, data_deteccao, regra, status, confianca, cidadao_cpf
-- Records: ~450K

3.2 Neo4j (graph — 13 databases)

cypher

-- Database: datta-datacatalog (default)
MATCH (n) RETURN labels(n) DISTINCT;
-- Nodes: Dataset, Column, Process, GlossaryTerm, LineageNode

-- Database: datta-ontology
MATCH (n) RETURN labels(n) DISTINCT;
-- Nodes: Ontology, EntityType, PropertyDef, DigitalTwin

-- Database: datta-graph
MATCH (n) RETURN labels(n) DISTINCT;
-- Nodes: Processo, Parte, Empresa, Decisao, Legislacao

3.3 OpenSearch (full text)

Three text indexes are available: processos, legislacao and documentos. You query them through the Python client already installed in the environment, as shown in Pattern 3 below.

4. Usage patterns

Pattern 1: distributive analysis (pure Trino)

When to use: massive aggregations, JOINs between large tables.

sql

SELECT 
    programa,
    COUNT(*) as qtd_beneficiarios,
    SUM(valor) as valor_total
FROM hive.assistencia_social.beneficios
GROUP BY programa
ORDER BY valor_total DESC;

Typical performance: ~2–5 seconds for 5.1M records.

Pattern 2: semantic analysis (Trino + Neo4j)

When to use: relationships, lineage, impact.

python

# 1. Extract data from Trino (SQL)
df_beneficiarios = spark.sql("""
    SELECT DISTINCT titular_cpf, programa 
    FROM hive.assistencia_social.beneficios
    LIMIT 1000
""")

# 2. Enrich with Neo4j (Cypher)
import os
from neo4j import GraphDatabase
driver = GraphDatabase.driver(
    os.environ.get("NEO4J_BOLT_URL", "bolt://neo4j:7687"),
    auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)
session = driver.session(database="datta-datacatalog")

cpfs = df_beneficiarios.select('titular_cpf').rdd.flatMap(list).collect()
query = """
MATCH (c:Cidadao {cpf: $cpf})-[:ENVOLVIDO_EM]->(p:Processo)
RETURN c.cpf, COUNT(p) as processos
"""
resultados = []
for cpf in cpfs[:100]:  # Sample
    result = session.run(query, cpf=cpf)
    resultados.extend([dict(r) for r in result])

Typical performance: ~500ms for 100 individual lookups — the same read done in a single batch can be up to 50 times faster.

Pattern 3: contextual discovery (OpenSearch + Trino)

When to use: finding the legislation relevant to a program.

python

from elasticsearch import Elasticsearch

es = Elasticsearch(["http://opensearch:9200"])

# Search legislation about a program
query = {
    "query": {
        "multi_match": {
            "query": "auxílio emergencial",
            "fields": ["titulo^2", "ementa^1.5", "conteudo"]
        }
    },
    "aggs": {
        "por_ano": {
            "date_histogram": {
                "field": "data_publicacao",
                "calendar_interval": "year"
            }
        }
    }
}

response = es.search(index="legislacao", body=query)

Typical performance: ~100–200ms (inverted index).

5. Practical example: integrated risk analysis

Goal

Identify beneficiaries with a risk pattern:

high income, yet eligible for a low-income program;
involved in a questionable judicial case;
unresolved alerts.

Complete flow

python

# 1. Trino: anomalous beneficiaries
df_anomalias = spark.sql("""
    SELECT 
        b.titular_cpf,
        b.programa,
        c.renda,
        b.valor,
        COUNT(a.id) as alertas_nao_resolvidos
    FROM hive.assistencia_social.beneficios b
    LEFT JOIN hive.assistencia_social.cidadaos c ON b.titular_cpf = c.cpf
    LEFT JOIN hive.assistencia_social.alertas a ON b.titular_cpf = a.cidadao_cpf 
        AND a.status != 'RESOLVIDO'
    WHERE c.renda > (SELECT PERCENTILE_CONT(0.75) WITHIN GROUP (ORDER BY renda) FROM hive.assistencia_social.cidadaos)
    GROUP BY b.titular_cpf, b.programa, c.renda, b.valor
    HAVING alertas_nao_resolvidos > 0
""")

# 2. Neo4j: enrich with judicial cases
from neo4j import GraphDatabase
driver = GraphDatabase.driver(
    os.environ.get("NEO4J_BOLT_URL", "bolt://neo4j:7687"),
    auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)
session = driver.session(database="datta-datacatalog")

anomalias_com_processos = []
for row in df_anomalias.collect():
    cpf = row.titular_cpf
    resultado = session.run("""
        MATCH (c:Cidadao {cpf: $cpf})-[:ENVOLVIDO_EM]->(p:Processo)
        WHERE p.status IN ['ATIVO', 'CONTESTADO']
        RETURN COUNT(p) as processos_ativos
    """, cpf=cpf)
    resultado_dict = {**row.asDict(), **dict(resultado.single())}
    anomalias_com_processos.append(resultado_dict)

# 3. OpenSearch: legislative context
legislacao_relevante = es.search(
    index="legislacao",
    body={
        "query": {
            "terms": {
                "programa": df_anomalias.select('programa').distinct().rdd.flatMap(list).collect()
            }
        }
    }
)

# 4. Pandas: final analysis
import pandas as pd
df_final = pd.DataFrame(anomalias_com_processos)
print(f"Beneficiários em risco: {len(df_final)}")
print(f"Valor total em risco: R$ {df_final['valor'].sum():.2f}")
print(f"Renda média: R$ {df_final['renda'].mean():.2f}")

At the end you have an auditable slice — who, how much and why — built from three sources in a single reproducible flow.

6. Performance and best practices

What to do

Pattern	Reason
`LIMIT 1000` during exploration	Avoids huge output
`SELECT col1, col2` (specific columns)	Parquet is columnar, selection is fast
`CAST(string_col AS DATE)` before GROUP BY	Avoids parsing the string N times
Reuse frequent values from the platform cache	Reduces repeated queries
`spark.sql(...).repartition(200)` before a join	Distributes the load

What to avoid

Pattern	Reason
`SELECT *`	Loads unnecessary columns
`JOIN` with `LIKE '%pattern%'`	Full scan, no index
Python UDFs in a loop	Expensive serialization
`collect()` without `LIMIT`	Can exhaust the session's memory
Blocking queries > 5min	Exceed the session's time limit

7. When something does not work

Symptom	What to do
Spark session not responding	In JupyterHub: `Kernel → Restart Kernel` and run the cells again. If it persists, ask the administrator to check the state and the resource limits of the notebook environment
`No such table: hive.assistencia_social.cidadaos`	Confirm with `SHOW TABLES IN hive.assistencia_social;`. If the list comes back empty, the sample database has not been loaded yet — ask the administrator to run the Trino seed (`k8s/trino/seed-trino.sql`)
`Neo4j connection timeout`	Check, with `os.environ`, that the `NEO4J_BOLT_URL`, `NEO4J_USERNAME` and `NEO4J_PASSWORD` variables reached the environment and that the graph answers on port 7687. Credentials come from the platform's secrets — never written into the notebook

8. Next steps

Run the example notebook → understand the patterns.
Create a custom query → for your use case.
Export results → to the catalog (Neo4j ProfilingSnapshot).
Publish the result → the Knowledge Catalog accepts dataset registration through a programmatic integration; see the API reference.
Schedule jobs → Apache Airflow (optional).

9. References

The environment runs on JupyterHub, reachable at /jupyter; the SQL engine is Trino 480 (replacing Impala) and the analytical storage is Parquet/ORC on HDFS + S3/GCS.

Guide: DATTA Multi-Source Notebook

Which source to use for each question

1. Opening and importing a notebook

Option A: through the interface (recommended)

Option B: through the notebooks folder

2. What changed: from Impala to Trino

Before (Impala)

Now (Trino)

Available catalogs

3. What exists in each source

3.1 Trino (Social Assistance)

3.2 Neo4j (graph — 13 databases)

3.3 OpenSearch (full text)

4. Usage patterns

Pattern 1: distributive analysis (pure Trino)

Pattern 2: semantic analysis (Trino + Neo4j)

Pattern 3: contextual discovery (OpenSearch + Trino)

5. Practical example: integrated risk analysis

Goal

Complete flow

6. Performance and best practices

What to do

What to avoid

7. When something does not work

8. Next steps

9. References