Using NiFi

Available in VPC

NiFi is a Dataflow engine, which is a stable system useful for handling and deploying data between different systems. NiFi is one of the ETL tools used to load the data after collecting and handling, and is also an open source that implements Flow-Based Programming (FBP) concept by collecting and handling vast data from a distributed environment.
NiFi is very suitable for real-time handling, having the advantage of data transfer without data loss.

NiFi components

NiFi is composed of FlowFile, Processor, Connector, and Controller. The following describes each component:

• FlowFile: Data unit recognized by NiFi.
  • Content: Data itself.
  • Attribute: Information related to the data and expressed as a key and value pair.
• Processor: Feature that collects, transforms, and stores FlowFile.
  • A new FlowFile can be created after completing data processing.
  • Multiple processors can operate in parallel.
• Connector: Connects processors to each other and transfers FlowFile.
  • Represents the queue of FlowFile.
  • Controls load by setting priorities and backpressure.
• Flow Controller: Connects each process and manages the FlowFile that passes between them.

Check NiFi service from Ambari UI

You can install NiFi using Add Service in the Ambari UI.

Options to enter during installation

• Replace the entire content of Advanced nifi-properties-env with the values below.
  Run the openssl rand command on the VM, copy the output value, and enter it for nifi.sensitive.props.key at the bottom of the content settings below.
  openssl rand -base64 32

# Core Properties #
nifi.flow.configuration.file=./conf/flow.xml.gz
nifi.flow.configuration.json.file=./conf/flow.json.gz
nifi.flow.configuration.archive.enabled=true
nifi.flow.configuration.archive.dir=./conf/archive/
nifi.flow.configuration.archive.max.time=30 days
nifi.flow.configuration.archive.max.storage=500 MB
nifi.flow.configuration.archive.max.count=
nifi.flowcontroller.autoResumeState=true
nifi.flowcontroller.graceful.shutdown.period=10 sec
nifi.flowservice.writedelay.interval=500 ms
nifi.administrative.yield.duration=30 sec
nifi.bored.yield.duration=10 millis
nifi.queue.backpressure.count=10000
nifi.queue.backpressure.size=1 GB

nifi.authorizer.configuration.file=./conf/authorizers.xml
nifi.login.identity.provider.configuration.file=./conf/login-identity-providers.xml
nifi.templates.directory=./conf/templates
nifi.ui.banner.text=
nifi.ui.autorefresh.interval=30 sec
nifi.nar.library.directory=./lib
nifi.nar.library.autoload.directory=./extensions
nifi.nar.working.directory=./work/nar/
nifi.documentation.working.directory=./work/docs/components
nifi.nar.unpack.uber.jar=false

####################
# State Management #
####################
nifi.state.management.configuration.file=./conf/state-management.xml
nifi.state.management.provider.local=local-provider
nifi.state.management.provider.cluster=zk-provider
nifi.state.management.embedded.zookeeper.start=false
nifi.state.management.embedded.zookeeper.properties=./conf/zookeeper.properties

nifi.database.directory=./database_repository
nifi.h2.url.append=;LOCK_TIMEOUT=25000;WRITE_DELAY=0;AUTO_SERVER=FALSE

nifi.repository.encryption.protocol.version=
nifi.repository.encryption.key.id=
nifi.repository.encryption.key.provider=
nifi.repository.encryption.key.provider.keystore.location=
nifi.repository.encryption.key.provider.keystore.password=

nifi.flowfile.repository.implementation=org.apache.nifi.controller.repository.WriteAheadFlowFileRepository
nifi.flowfile.repository.wal.implementation=org.apache.nifi.wali.SequentialAccessWriteAheadLog
nifi.flowfile.repository.directory=./flowfile_repository
nifi.flowfile.repository.checkpoint.interval=20 secs
nifi.flowfile.repository.always.sync=false
nifi.flowfile.repository.retain.orphaned.flowfiles=true

nifi.swap.manager.implementation=org.apache.nifi.controller.FileSystemSwapManager
nifi.queue.swap.threshold=20000

nifi.content.repository.implementation=org.apache.nifi.controller.repository.FileSystemRepository
nifi.content.claim.max.appendable.size=50 KB
nifi.content.repository.directory.default=./content_repository
nifi.content.repository.archive.max.retention.period=7 days
nifi.content.repository.archive.max.usage.percentage=50%
nifi.content.repository.archive.enabled=true
nifi.content.repository.always.sync=false
nifi.content.viewer.url=../nifi-content-viewer/

nifi.provenance.repository.implementation=org.apache.nifi.provenance.WriteAheadProvenanceRepository
nifi.provenance.repository.directory.default=./provenance_repository
nifi.provenance.repository.max.storage.time=30 days
nifi.provenance.repository.max.storage.size=10 GB
nifi.provenance.repository.rollover.time=10 mins
nifi.provenance.repository.rollover.size=100 MB
nifi.provenance.repository.query.threads=2
nifi.provenance.repository.index.threads=2
nifi.provenance.repository.compress.on.rollover=true
nifi.provenance.repository.always.sync=false
nifi.provenance.repository.indexed.fields=EventType, FlowFileUUID, Filename, ProcessorID, Relationship
nifi.provenance.repository.indexed.attributes=
nifi.provenance.repository.index.shard.size=500 MB
nifi.provenance.repository.max.attribute.length=65536
nifi.provenance.repository.concurrent.merge.threads=2

nifi.provenance.repository.buffer.size=100000

nifi.components.status.repository.implementation=org.apache.nifi.controller.status.history.VolatileComponentStatusRepository
nifi.components.status.repository.buffer.size=1440
nifi.components.status.snapshot.frequency=1 min
nifi.status.repository.questdb.persist.node.days=14
nifi.status.repository.questdb.persist.component.days=3
nifi.status.repository.questdb.persist.location=./status_repository

nifi.remote.input.host=
nifi.remote.input.secure=false
nifi.remote.input.socket.port=
nifi.remote.input.http.enabled=true
nifi.remote.input.http.transaction.ttl=30 sec
nifi.remote.contents.cache.expiration=30 secs

nifi.web.http.host=0.0.0.0
nifi.web.http.port={{nifi_port}}
nifi.web.http.network.interface.default=

nifi.web.https.host=
nifi.web.https.port=
nifi.web.https.network.interface.default=
nifi.web.https.application.protocols=http/1.1
nifi.web.jetty.working.directory=./work/jetty
nifi.web.jetty.threads=200
nifi.web.max.header.size=16 KB
nifi.web.proxy.context.path=
nifi.web.proxy.host=
nifi.web.max.content.size=
nifi.web.max.requests.per.second=30000
nifi.web.max.access.token.requests.per.second=25
nifi.web.request.timeout=60 secs
nifi.web.request.ip.whitelist=
nifi.web.should.send.server.version=true
nifi.web.request.log.format=%{client}a - %u %t "%r" %s %O "%{Referer}i" "%{User-Agent}i"

nifi.sensitive.props.key= [Enter OPENSSL RAND value]
nifi.sensitive.props.algorithm=NIFI_PBKDF2_AES_GCM_256

• For Secure Hadoop, add the following authentication information to the content values above, and replace the content of Advanced nifi-properties-env with the resulting values.

nifi.security.autoreload.enabled=false
nifi.security.autoreload.interval=10 secs
nifi.security.user.authorizer=single-user-authorizer
nifi.security.allow.anonymous.authentication=false
nifi.security.user.login.identity.provider=single-user-provider
nifi.security.user.jws.key.rotation.period=PT1H
nifi.kerberos.krb5.file=/etc/krb5.conf
nifi.kerberos.service.principal=nifi/_HOST@realm_information
nifi.kerberos.service.keytab.location=/etc/security/keytabs/nifi.keytab
nifi.kerberos.spnego.principal=HTTP/_HOST@realm_information
nifi.kerberos.spnego.keytab.location=/etc/security/keytabs/spnego.service.keytab

Using NiFi

The following describes how to create Data Flow to transfer the local file to HDFS: The following describes the complete process and step-by-step explanation of how files created in the local directory are transferred to HDFS through NiFi Data Flow:

1. Create GetFile processor
2. Create local file
3. Create HDFS processor
4. Connect processor
5. Run and view result
6. Troubleshooting

1. Create GetFile processor

To create a GetFile processor in NiFi:

1. Create a nifi-test directory in the master node's /tmp directory, as shown below.

mkdir /tmp/nifi-test
chown nifi /tmp/nifi-test

2. Drag the processor from the NiFi Web GUI component toolbar to the canvas.
   

3. Create and set the GetFile processor in the NiFi Web GUI.
   

4. Right-click the GetFile processor and click [Configure].

5. Click the [PROPERTIES] tab.

6. Enter the location of the nifi-test directory from the input directory list and click [APPLY].
   

2. Create local file

To create a temporary file in a local environment: Create a test1.txt file using vi as shown below.

[irteamsu@dev-nch271-ncl nifi-test]$ vi test1.txt

3. Create HDFS processor

To create an HDFS processor in NiFi:

1. Create an HDFS directory for the data to be saved in the local environment as shown below.

sudo -u hdfs hdfs dfs -mkdir -p /user/nifi
sudo -u hdfs hdfs dfs -chown nifi /user/nifi

2. Create a PutHDFS processor on the canvas through the NiFi GUI in the NiFi Web GUI.
3. Right-click the PutHDFS processor and click [Configure].
4. Click the [PROPERTIES] tab and enter the property as shown below.

• Hadoop Configuration Resources: /etc/hadoop/conf/core-site.xml,/etc/hadoop/conf/hdfs-site.xml
• Directory: /user/nifi

5. Click the [RELATIONSHIPS] tab and check all of the Terminate from Failure and Success.
   

4. Connect processor

To connect the PutHDFS processor and the GetFile processor in NiFi:

1. Move the mouse cursor to the GetFile processor in the NiFi Web GUI and drag the connect icon to connect to the PutHDFS processor.
2. The complete Data Flow is shown below:
   

5. Run and view result

To check the NiFi Data Flow running in a local environment and the result:

1. Access the NiFi Web GUI.
2. Right-click the GetFile processor, and click [Start].
3. Right-click the PutHDFS processor and click [Start].
4. To check the /tmp/nifi-test directory, enter the following in the local environment:

   [irteamsu@dev-nch271-ncl nifi-test]$ pwd
    /tmp/nifi-test
   [irteamsu@dev-nch271-ncl nifi-test]$ ls
   

   • The file from the local environment was transferred to HDFS through NiFi Data Flow, so the file from /tmp/nifi-test directory does not exist.
5. To check the files transferred to HDFS, enter the following commands in your local environment:

   [irteamsu@dev-nch271-ncl nifi-test]$ sudo -u hdfs hdfs dfs -ls /user/nifi
   Found 1 items
   -rw-r--r--   2 nifi hdfs          4 2023-08-29 17:33 /user/nifi/test1.txt
   [irteamsu@dev-nch271-ncl nifi-test]$ sudo -u hdfs hdfs dfs -cat /user/nifi/test1.txt   
   

   • The file from the local environment was transferred to HDFS through NiFi Data Flow.

6. Troubleshoot

When a problem occurs in Data Flow, you can troubleshoot by checking Data Provenance. To check Data Provenance:

1. Right-click the processor in the NiFi Web GUI to check.
2. Click [View data provenance].
3. You can see the Data Provenance information below.