Apache NiFi

# Download e instalação
wget https://downloads.apache.org/nifi/1.24.0/nifi-1.24.0-bin.tar.gz
tar -xzf nifi-1.24.0-bin.tar.gz
cd nifi-1.24.0

# Configurar Java (se necessário)
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk

# Iniciar NiFi
./bin/nifi.sh start

# Verificar status
./bin/nifi.sh status

# Acessar interface web
# http://localhost:8080/nifi

# Web properties
nifi.web.http.host=localhost
nifi.web.http.port=8080
nifi.web.https.port=8443

# Cluster properties
nifi.cluster.is.node=false
nifi.cluster.node.address=localhost
nifi.cluster.node.protocol.port=11443

# Security properties
nifi.security.keystore=
nifi.security.keystoreType=
nifi.security.keystore.passwd=
nifi.security.truststore=
nifi.security.truststoreType=
nifi.security.truststore.passwd=

# Repository properties
nifi.flowfile.repository.directory=./flowfile_repository
nifi.content.repository.directory.default=./content_repository
nifi.provenance.repository.directory.default=./provenance_repository

# Performance tuning
nifi.flowcontroller.graceful.shutdown.period=10 sec
nifi.flowservice.writedelay.interval=500 ms
nifi.administrative.yield.duration=30 sec

Fluxo Visual no NiFi:

1. GetFile Processor
   - Directory: /input
   - File Filter: *.json
   - Keep Source File: false

2. EvaluateJsonPath Processor
   - Destination: flowfile-attribute
   - user_id: $.user_id
   - event_type: $.event_type

3. RouteOnAttribute Processor
   - Route based on event_type
   - login_events: ${event_type:equals('login')}
   - purchase_events: ${event_type:equals('purchase')}

4. PublishKafka_2_6 Processor
   - Kafka Brokers: localhost:9092
   - Topic Name: user-events
   - Message Key Field: user_id

@Tags({"custom", "transform", "example"})
@CapabilityDescription("Custom processor example")
public class CustomTransformProcessor extends AbstractProcessor {
    
    public static final PropertyDescriptor TRANSFORM_TYPE = new PropertyDescriptor.Builder()
        .name("Transform Type")
        .description("Type of transformation to apply")
        .required(true)
        .allowableValues("uppercase", "lowercase", "reverse")
        .defaultValue("uppercase")
        .build();
    
    public static final Relationship REL_SUCCESS = new Relationship.Builder()
        .name("success")
        .description("Successfully transformed FlowFiles")
        .build();
    
    public static final Relationship REL_FAILURE = new Relationship.Builder()
        .name("failure")
        .description("Failed to transform FlowFiles")
        .build();
    
    @Override
    protected List getSupportedPropertyDescriptors() {
        return Arrays.asList(TRANSFORM_TYPE);
    }
    
    @Override
    public Set getRelationships() {
        return new HashSet<>(Arrays.asList(REL_SUCCESS, REL_FAILURE));
    }
    
    @Override
    public void onTrigger(ProcessContext context, ProcessSession session) {
        FlowFile flowFile = session.get();
        if (flowFile == null) {
            return;
        }
        
        try {
            String transformType = context.getProperty(TRANSFORM_TYPE).getValue();
            
            flowFile = session.write(flowFile, (inputStream, outputStream) -> {
                String content = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
                String transformed;
                
                switch (transformType) {
                    case "uppercase":
                        transformed = content.toUpperCase();
                        break;
                    case "lowercase":
                        transformed = content.toLowerCase();
                        break;
                    case "reverse":
                        transformed = new StringBuilder(content).reverse().toString();
                        break;
                    default:
                        transformed = content;
                }
                
                outputStream.write(transformed.getBytes(StandardCharsets.UTF_8));
            });
            
            session.transfer(flowFile, REL_SUCCESS);
            
        } catch (Exception e) {
            getLogger().error("Failed to transform FlowFile", e);
            session.transfer(flowFile, REL_FAILURE);
        }
    }
}

version: '3.8'
services:
  nifi:
    image: apache/nifi:1.24.0
    container_name: nifi
    ports:
      - "8080:8080"
      - "8443:8443"
    environment:
      - NIFI_WEB_HTTP_PORT=8080
      - NIFI_CLUSTER_IS_NODE=true
      - NIFI_CLUSTER_NODE_PROTOCOL_PORT=8082
      - NIFI_ZK_CONNECT_STRING=zookeeper:2181
      - NIFI_ELECTION_MAX_WAIT=30 sec
      - NIFI_SENSITIVE_PROPS_KEY=changemeplease
    volumes:
      - nifi_database_repository:/opt/nifi/nifi-current/database_repository
      - nifi_flowfile_repository:/opt/nifi/nifi-current/flowfile_repository
      - nifi_content_repository:/opt/nifi/nifi-current/content_repository
      - nifi_provenance_repository:/opt/nifi/nifi-current/provenance_repository
      - nifi_state:/opt/nifi/nifi-current/state
      - nifi_logs:/opt/nifi/nifi-current/logs
      - ./input:/input
      - ./output:/output
    networks:
      - nifi_network

  zookeeper:
    image: zookeeper:3.7
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOO_MY_ID: 1
      ZOO_SERVERS: server.1=0.0.0.0:2888:3888;2181
    networks:
      - nifi_network

volumes:
  nifi_database_repository:
  nifi_flowfile_repository:
  nifi_content_repository:
  nifi_provenance_repository:
  nifi_state:
  nifi_logs:

networks:
  nifi_network:
    driver: bridge