Apache NiFi 로 데이터 흐름 제어하기

apache nifi, elasticsearch, kibana 모두 처음 다뤄보는 기술이라 잘 못된 설명이 있을 수 있습니다.

Why

저희 팀은 모니터링 데이터를 수집하고 메신저로 전달하여 실시간으로 정보를 받을 수 있는 서비스를 운영 중입니다.

문제는 모든 데이터를 항상 메신저로 보내주고 있다는 점이며 만약 갑작스런 오류로 대량의 데이터가 수집되거나 중요한 데이터가 저장되지 않아 다시 찾아 볼 수 없는 상황입니다.

저희는 기민하게 움직여야 하는 조직이며 업무와 병행하며 다양한 이슈를 처리하고 있어 한 사람이 부족한 상황에서 데이터 흐름을 그때그때 변경 하기란 간단한 일이 아닙니다. 사실 귀찮은 작업이기도 합니다. :)

그래서...

모든 데이터는 수집되어야 하고 원하는 흐름을 쉽게 설계할 수 있어야 한다.

생각했습니다.

제 생각에 걸 맞는 오픈 프로젝트 Apache NiFi 를 찾고 한번 원하는 설계를 재현해 보기로 했습니다.

What

직관적인 UI 로 누구나 쉽게 데이터 흐름을 설계할 수 있으며 실시간으로 데이터 흐름을 차단하거나 새로운 플랫폼에 데이터를 전달할 수 있는 아주 훌륭한 도구입니다.

이제...

모든 데이터는 수집하고 원하는 흐름을 쉽게 설계할 수 있습니다.

• Rabbitmq 발행 및 수집 (Kafka 지원)
• elasticsearch 데이터 저장과 kibana 데이터 시각화
• slack 메신저로 메세지 출력 (HTTP, SFTP, FTP 왠만한 프로토콜 지원)
• Text 데이터 출력 가능
• Json 데이터 추출 출력 가능
• 그외 다양한 프로세서를 지원

결과 영상

How

• 메세지 브로커로 수집된 데이터 슬랙으로 보내기
• 메세지 브로커로 수집된 데이터 elasticsearch(kibana 로 확인) 와 슬랙으로 보내기
• 데이터를 텍스트로 출력하기
• 데이터를 JSON 으로 파싱하여 출력하기

위 구성을 재현하기 위해 도커를 사용했습니다.

 Apache NiFi

    version: '3.3'
    services:
      nifi:
        image: 'apache/nifi:latest'
        container_name: 'nifi'
        hostname: 'nifi'
        ports:
          - '8090:8080'
    # 혹시 볼륨으로 설정하는 법 아시나요?
    #    volumes:
    #      - './nifi:/opt/nifi/nifi-current'
    networks:
      default:
        external:
          name: devops

RabbitMQ Docker Compose

    version: '3'

    services:
      rabbitmq:
        image: 'rabbitmq:3-management'
        container_name: 'rabbitmq'
        hostname: 'rabbitmq'
        ports:
          - '15672:15672'
          - '5672:5672'
        volumes:
          - './rabbitmq_data:/var/lib/rabbitmq'
        environment:
          RABBITMQ_DEFAULT_USER: 'local'
          RABBITMQ_DEFAULT_PASS: '123!@#123fhzjf'
    networks:
      default:
        external:
          name: devops

ElasticSearch + Kibana

    version: '2.2'
    services:
      es01:
        image: docker.elastic.co/elasticsearch/elasticsearch:7.3.0
        container_name: es01
        environment:
          - node.name=es01
          - discovery.seed_hosts=es02
          - cluster.initial_master_nodes=es01,es02
          - cluster.name=docker-cluster
          - bootstrap.memory_lock=true
          - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
        ulimits:
          memlock:
            soft: -1
            hard: -1
        volumes:
          - esdata01:/usr/share/elasticsearch/data
        ports:
          - 9200:9200
        networks:
          - default
      es02:
        image: docker.elastic.co/elasticsearch/elasticsearch:7.3.0
        container_name: es02
        environment:
          - node.name=es02
          - discovery.seed_hosts=es01
          - cluster.initial_master_nodes=es01,es02
          - cluster.name=docker-cluster
          - bootstrap.memory_lock=true
          - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
        ulimits:
          memlock:
            soft: -1
            hard: -1
        volumes:
          - esdata02:/usr/share/elasticsearch/data
        networks:
          - default
      kibana:
        image: docker.elastic.co/kibana/kibana:7.3.0
        container_name: kibana
        environment:
          ELASTICSEARCH_HOSTS: http://es01:9200
        ports:
          - 5601:5601
        networks:
          - default
        depends_on:
          - es01
          - es02
    volumes:
      esdata01:
        driver: local
      esdata02:
        driver: local
    networks:
      default:
        external:
          name: devops

실행하기

docker start es01 es02 kibana rabbitmq nifi

• nifi: http://localhost:8090
• kibana: http://localhost:5601
• rabbitmq: http://localhost:15672

데이터 흐름 설계

다양한 형식을 사용해 보기 위해 불필요한 부분도 있습니다.

테스트를 위한 프로세서