Apache NiFi: Como agrupar e transformar dados do SQL Server em JSON com Jolt?

Apache NiFi: Como agrupar e transformar dados do SQL Server em JSON com o Jolt

Aprenda como agrupar e transoformar dados do SQL Server em JSON com o Jolt utilizando o Apache NiFi. Apache NiFi é uma solução de integração de dados e de sistemas e um grande facilitador para lidar como estratégia para modernização de software legado que fornece um ambiente visual para desenvolver dataflows. Ele facilita a transformação e a integração de dados provenientes de várias fontes e para vários destinos. Uma das suas características mais potentes é a habilidade de transformar dados on-the-fly usando a linguagem de transformação Jolt.

Aplicando Enterprise Integration Pattern na prática

Apache NiFi implementa vários padrões de integração empresarial – Enterprise Integration Patterns – como Message Routing, Message Transformation, etc. Este artigo focará principalmente no padrão de Message Transformation, onde os dados extraídos de um sistema são transformados em um formato adequado para um outro sistema.

Construíndo nosso ambiente de testes com SQL Server e Apache NiFi

Antes de mergulharmos nas etapas específicas de como usar o Apache NiFi para agrupar e transformar dados do SQL Server, precisamos ter um ambiente de testes para experimentar. Este ambiente simula um caso real de um banco de dados SQL Server contendo informações de clientes e seus respectivos contatos. Além disso, utilizaremos o Apache NiFi para orquestrar e transformar esses dados.

Para simplificar o processo de configuração, vamos utilizar o Docker para criar nosso ambiente de testes. Isso permitirá que qualquer pessoa possa recriar o mesmo ambiente de forma rápida e sem complicações.

version: '3.4'

services:
  mssql-server:    
    image: "mcr.microsoft.com/mssql/server:2019-latest"
    environment:
      ACCEPT_EULA: "Y"    
      MSSQL_SA_PASSWORD: "s3nhaEhPa55w0rD"
      MSSQL_COLLATION: "SQL_Latin1_General_CP1_CI_AS"
      MSSQL_AGENT_ENABLED: "True"
    ports:
      - "1433:1433"
    container_name: "mssql-server"
    volumes:
      - .docker/volume/sqlserver/data:/var/opt/mssql/data
      - .docker/volume/sqlserver/log:/var/opt/mssql/log
      - .docker/volume/sqlserver/secrets:/var/opt/mssql/secrets
    restart: always      
    networks:
      - demo  

  nifi:
    image: apache/nifi
    container_name: "nifi"
    environment:
      - NIFI_WEB_HTTP_PORT=8080
    ports:
      - 8080:8080
    restart: always      
    networks:
      - demo

networks:
  demo:
    driver: bridge

version: '3.4'

services:

mssql-server:

image: "mcr.microsoft.com/mssql/server:2019-latest"

environment:

ACCEPT_EULA: "Y"

MSSQL_SA_PASSWORD: "s3nhaEhPa55w0rD"

MSSQL_COLLATION: "SQL_Latin1_General_CP1_CI_AS"

MSSQL_AGENT_ENABLED: "True"

ports:

- "1433:1433"

container_name: "mssql-server"

volumes:

- .docker/volume/sqlserver/data:/var/opt/mssql/data

- .docker/volume/sqlserver/log:/var/opt/mssql/log

- .docker/volume/sqlserver/secrets:/var/opt/mssql/secrets

restart: always

networks:

- demo

nifi:

image: apache/nifi

container_name: "nifi"

environment:

- NIFI_WEB_HTTP_PORT=8080

ports:

- 8080:8080

restart: always

networks:

- demo

networks:

demo:

driver: bridge

Construindo nosso ambiente de testes com SQL Server e Apache NiFi

Antes de mergulharmos nas etapas específicas de como usar o Apache NiFi para agrupar e transformar dados do SQL Server, é fundamental ter um ambiente de testes para experimentar. Este ambiente simula um caso real de um banco de dados SQL Server contendo informações de clientes e seus respectivos contatos. Além disso, utilizaremos o Apache NiFi para orquestrar e transformar esses dados.

Ambiente Dockerizado

Código docker-compose:

version: '3.4'

services:
  mssql-server:    
    image: "mcr.microsoft.com/mssql/server:2019-latest"
    environment:
      ACCEPT_EULA: "Y"    
      MSSQL_SA_PASSWORD: "s3nhaEhPa55w0rD"
      MSSQL_COLLATION: "SQL_Latin1_General_CP1_CI_AS"
      MSSQL_AGENT_ENABLED: "True"
    ports:
      - "1433:1433"
    container_name: "mssql-server"
    volumes:
      - .docker/volume/sqlserver/data:/var/opt/mssql/data
      - .docker/volume/sqlserver/log:/var/opt/mssql/log
      - .docker/volume/sqlserver/secrets:/var/opt/mssql/secrets
    restart: always      
    networks:
      - demo  

  nifi:
    image: apache/nifi
    container_name: "nifi"
    environment:
      - NIFI_WEB_HTTP_PORT=8080
    ports:
      - 8080:8080
    restart: always      
    networks:
      - demo

networks:
  demo:
    driver: bridge

version: '3.4'

services:

mssql-server:

image: "mcr.microsoft.com/mssql/server:2019-latest"

environment:

ACCEPT_EULA: "Y"

MSSQL_SA_PASSWORD: "s3nhaEhPa55w0rD"

MSSQL_COLLATION: "SQL_Latin1_General_CP1_CI_AS"

MSSQL_AGENT_ENABLED: "True"

ports:

- "1433:1433"

container_name: "mssql-server"

volumes:

- .docker/volume/sqlserver/data:/var/opt/mssql/data

- .docker/volume/sqlserver/log:/var/opt/mssql/log

- .docker/volume/sqlserver/secrets:/var/opt/mssql/secrets

restart: always

networks:

- demo

nifi:

image: apache/nifi

container_name: "nifi"

environment:

- NIFI_WEB_HTTP_PORT=8080

ports:

- 8080:8080

restart: always

networks:

- demo

networks:

demo:

driver: bridge

Este script Docker-Compose configura dois serviços:

mssql-server: Um contêiner rodando o SQL Server 2019.
nifi: Um contêiner rodando o Apache NiFi na porta 8080.

Após iniciar o contêiner do SQL Server, o próximo passo é criar as tabelas Customer e Contact.

GO
CREATE TABLE [Customer]
(
   [id]     INT IDENTITY(1,1),
   [name]   NVARCHAR(75) NOT NULL
)
GO
ALTER TABLE [Customer] ADD CONSTRAINT [PK_Customer] PRIMARY KEY
(
   [id]
)
GO
CREATE TABLE [Contact]
(
   [id]         INT IDENTITY(1,1), 
   [customerId] INT NOT NULL,
   [type]       INT NOT NULL,
   [phone]      NVARCHAR(11) NOT NULL 
)
GO
ALTER TABLE [Contact] ADD CONSTRAINT [PK_Contact] PRIMARY KEY
(
   [id]
)
GO
CREATE INDEX [IX_Contact_001] ON [Contact]([customerId]) 
GO
ALTER TABLE [Contact] ADD CONSTRAINT [FK_Contact_Customer] FOREIGN KEY ([customerId]) REFERENCES [Customer]([id])
GO

CREATE TABLE [Customer]

(

[id] INT IDENTITY(1,1),

[name] NVARCHAR(75) NOT NULL

)

ALTER TABLE [Customer] ADD CONSTRAINT [PK_Customer] PRIMARY KEY

(

[id]

)

CREATE TABLE [Contact]

(

[id] INT IDENTITY(1,1),

[customerId] INT NOT NULL,

[type] INT NOT NULL,

[phone] NVARCHAR(11) NOT NULL

)

ALTER TABLE [Contact] ADD CONSTRAINT [PK_Contact] PRIMARY KEY

(

[id]

)

CREATE INDEX [IX_Contact_001] ON [Contact]([customerId])

ALTER TABLE [Contact] ADD CONSTRAINT [FK_Contact_Customer] FOREIGN KEY ([customerId]) REFERENCES [Customer]([id])

A seguir, populamos as tabelas com dados de exemplo:

INSERT INTO Customer([name])VALUES('Nome do Cliente');
INSERT INTO Contact([customerId],[type],[phone])VALUES(1,1,'11999999993')
INSERT INTO Contact([customerId],[type],[phone])VALUES(1,1,'11987654321')

INSERT INTO Customer([name])VALUES('Nome do Cliente');

INSERT INTO Contact([customerId],[type],[phone])VALUES(1,1,'11999999993')

INSERT INTO Contact([customerId],[type],[phone])VALUES(1,1,'11987654321')

Finalmente, uma consulta SQL junta as informações do cliente e seus contatos:

SELECT  cust.[id], cust.[name], cont.[type], cont.[phone] 
FROM    Customer cust WITH(NOLOCK)
JOIN    Contact  cont WITH(NOLOCK) ON cust.id = cont.customerId
WHERE   cust.id = 1

SELECT cust.[id], cust.[name], cont.[type], cont.[phone]

FROM Customer cust WITH(NOLOCK)

JOIN Contact cont WITH(NOLOCK) ON cust.id = cont.customerId

WHERE cust.id = 1

Agora que temos nosso ambiente de testes configurado, vamos para as etapas de implementação do fluxo de dados no Apache NiFi.

Criando o fluxo do Apache NiFi

Depois de termos o nosso ambiente de testes pronto e a consulta SQL para puxar os dados do SQL Server, a próxima etapa é criar um fluxo de dados no Apache NiFi para orquestrar todo o processo.

GenerateFlowFile

O primeiro processador que vamos utilizar é o GenerateFlowFile. Este processador é usado para criar um arquivo de fluxo inicial que dispara o restante do fluxo. No nosso caso, ele atua como um disparador para a ação de buscar os dados do SQL Server.

Database Connection Pooling Service

Para que o NiFi possa se comunicar com o SQL Server, precisamos configurar um serviço de agrupamento de conexões. Isso permite que várias operações de banco de dados reutilizem as mesmas conexões, melhorando a eficiência e a performance.

ExecuteSQL

Após configurar a conexão, utilizamos o processador ExecuteSQL, que executa a consulta SQL para extrair os dados. Este processador utiliza o serviço de agrupamento de conexões que configuramos anteriormente.

Apache NiFi: Como Agrupar e Transformar Dados do SQL Server em JSON com Jolt

ConvertAvroToJson

Depois de ter os dados, é hora de transformá-los em um formato mais acessível. O processador ConvertAvroToJson converte os dados de Avro para JSON, permitindo manipulações mais simples no próximo estágio.

JoltTransformJSON

Finalmente, chegamos ao processador JoltTransformJSON. Este é o estágio onde efetivamente transformamos os dados para se adequar ao formato desejado. Jolt é uma biblioteca que permite a transformação de dados JSON de maneira declarativa. No nosso caso, usamos Jolt para transformar e agrupar os dados dos clientes e contatos em um único JSON.

[
  {
    "operation": "shift",
    "spec": {
      "0": {
        "id": "id",
        "name": "name",
        "type": "contacts[].type",
        "phone": "contacts[].phone"
      },
      "*": {
        "type": "contacts[].type",
        "phone": "contacts[].phone"
      }
    }
  }
]

[

{

"operation": "shift",

"spec": {

"0": {

"id": "id",

"name": "name",

"type": "contacts[].type",

"phone": "contacts[].phone"

"*": {

"type": "contacts[].type",

"phone": "contacts[].phone"

}

]

Relacionando processors do Apache NiFi com Enterprise Integration Patterns

Enterprise Integration Patterns (EIP) são arquiteturas comprovadas para abordar problemas específicos na integração entre sistemas heterogêneos. Esses padrões fornecem um guia para construir sistemas de integração robustos, escaláveis e sustentáveis. Em nosso fluxo de dados com o Apache NiFi, vários desses padrões foram empregados, garantindo uma solução eficaz e resiliente.

Message Source e Message Channel

Neste contexto, o GenerateFlowFile e o ExecuteSQL funcionam como nossas fontes de mensagens (Message Source). Eles iniciam o fluxo de dados, emitindo eventos que são passados através de canais de mensagens (Message Channels), como filas ou tópicos, para os processadores seguintes.

Message Source: É o ponto inicial de onde as mensagens são produzidas. No nosso caso, o GenerateFlowFile age como tal, disparando o fluxo de dados.
Message Channel: É o “tubo” pelo qual a mensagem viaja de um ponto a outro. No Apache NiFi, isso é mais abstracto mas é essencialmente o que acontece quando ligamos processadores através de suas relações.

Message Transformation

Nossos processadores ConvertAvroToJson e JoltTransformJSON são responsáveis pela transformação das mensagens. Essa é uma implementação do padrão Message Transformation, que sugere que uma mensagem extraída de um sistema pode precisar ser transformada de alguma forma para ser útil para outro sistema.

Content Enricher: Aqui, o processador ExecuteSQL poderia ser considerado um enriquecedor de dados, pois ele junta informações de diferentes tabelas em um único conjunto de dados.
Translator: ConvertAvroToJson e JoltTransformJSON atuam como tradutores, convertendo o formato da mensagem de Avro para JSON e então transformando esse JSON para se adequar às nossas necessidades específicas.

Conclusão

Integrar e transformar dados pode parecer uma tarefa complexa, mas com as ferramentas certas e os padrões apropriados, é algo completamente acessível. Utilizando o Apache NiFi em conjunto com os Enterprise Integration Patterns, abrimos um mundo de possibilidades para manipulação de dados de forma eficaz, escalável e sustentável. Esperamos que este artigo tenha sido um guia prático para você, oferecendo o conhecimento e as etapas necessárias para começar a resolver seus próprios desafios de integração de dados.

FAQ: Perguntas Frequentes

1. O que são Enterprise Integration Patterns (EIP) e por que são importantes?

Enterprise Integration Patterns são arquiteturas padrão usadas para resolver problemas comuns na integração de sistemas heterogêneos. Eles são importantes porque fornecem um roteiro comprovado para construir sistemas de integração robustos, escaláveis e sustentáveis.

2. Por que usar Apache NiFi para a integração de dados?

O Apache NiFi oferece uma interface gráfica intuitiva, facilitando a construção de fluxos de dados complexos sem exigir extenso conhecimento em programação. Ele é altamente extensível e compatível com uma grande variedade de fontes de dados, tornando-o uma excelente opção para a integração de dados.

3. Como o Apache NiFi se compara a outras ferramentas de integração de dados?

O Apache NiFi é conhecido por sua flexibilidade e pela facilidade de uso, permitindo a criação rápida de fluxos de dados. Ele também se destaca em cenários onde o fluxo de dados em tempo real é necessário.

4. Quais são os desafios comuns ao agrupar dados de diferentes fontes e como o Apache NiFi ajuda a resolver esses problemas?

O agrupamento de dados de diferentes fontes muitas vezes envolve a resolução de problemas como formatos de dados inconsistentes, ordens de execução e questões de sincronização. O Apache NiFi ajuda a resolver esses problemas oferecendo uma ampla gama de processadores para transformação de dados e recursos para orquestração do fluxo de dados.

5. É seguro usar o Apache NiFi em um ambiente empresarial?

Sim, o Apache NiFi foi construído com foco em segurança e oferece várias funcionalidades, como controle de acesso baseado em funções, criptografia e auditoria, tornando-o apropriado para uso em ambientes empresariais com requisitos rigorosos de segurança.

Compartilhe:

Tiago Tartari

Eu ajudo e capacito pessoas e organizações a transformar problemas complexos em soluções práticas usando a tecnologia para atingir resultados extraordinários.

Qual é o desafio
que você tem hoje?

Apache NiFi: Como agrupar e transformar dados do SQL Server em JSON com Jolt?

Conteúdo

Apache NiFi: Como agrupar e transformar dados do SQL Server em JSON com o Jolt

Aplicando Enterprise Integration Pattern na prática

Construíndo nosso ambiente de testes com SQL Server e Apache NiFi

Construindo nosso ambiente de testes com SQL Server e Apache NiFi

Ambiente Dockerizado

Código docker-compose:

Criando o fluxo do Apache NiFi

GenerateFlowFile

Database Connection Pooling Service

ExecuteSQL

ConvertAvroToJson

JoltTransformJSON

Relacionando processors do Apache NiFi com Enterprise Integration Patterns

Message Source e Message Channel

Message Transformation

Conclusão

FAQ: Perguntas Frequentes

1. O que são Enterprise Integration Patterns (EIP) e por que são importantes?

2. Por que usar Apache NiFi para a integração de dados?

3. Como o Apache NiFi se compara a outras ferramentas de integração de dados?

4. Quais são os desafios comuns ao agrupar dados de diferentes fontes e como o Apache NiFi ajuda a resolver esses problemas?

5. É seguro usar o Apache NiFi em um ambiente empresarial?