ijokarumawak

NiFi Loop flow example

This template is analogous to the traditional for(i = 0; i < x; i++) loop in NiFi Data flow.

ijokarumawak

How to use Dante SOCKS proxy server

Create sockd.conf file locally on the docker host machine and map it with -v option of docker run.

Run socks server

docker run -d -p 1080:1080 -v /etc/sockd.conf:/etc/sockd.conf --name socks-server vimagick/dante

Add user

ijokarumawak

This example demonstrates Nginx reverse proxy configurations. If the client nor Nginx does NOT provide any client certificate, NiFi will respond with a login screen.

Nifi has to be configured to use an identity provider for username/password login. This blog post might be helpful to do so, it uses LDAP authentication provider. http://ijokarumawak.github.io/nifi/2016/11/15/nifi-auth/#how-to-use-ldap-to-authn-requests

ijokarumawak

NiFi example on how to join CSV files to create a merged result

Let's say there're following 3 CSV files (a, b and c):

t, v
1, 10
2, 20
3, 30
4, 40

ijokarumawak

NiFi Flow Example: Move Files between S3 Buckets

Please be careful with specifying the right bucket name, region and credential.

When I misconfigured region, I got the following error:

2016-11-17 10:51:07,828 ERROR [Timer-Driven Process Thread-9] o.a.nifi.processors.aws.s3.PutS3Object
com.amazonaws.services.s3.model.AmazonS3Exception: The bucket you are attempting to access must be addressed using the specified endpoint. Please send all future requests to this endpoint. (Service: Amazon S3; Status Code: 301; ErrorCode: PermanentRedirect; Request ID: 99A62426D8544997)

ijokarumawak

<?xml version="1.0" ?>
<template encoding-version="1.1">
  <description></description>
  <groupId>1e8778a8-015c-1000-c747-8273276bab45</groupId>
  <name>JSON Array to String List</name>
  <snippet>
    <processGroups>
      <id>51eb9288-0a70-311b-0000-000000000000</id>
      <parentGroupId>5454b790-94fe-3e28-0000-000000000000</parentGroupId>
      <position>

ijokarumawak

NiFi Example: Load CSV file into RDBMS Table using the traditional way and the new way using Record

Example Data

Ingested to the example flows by GenerateFlowFile:

ID, CITY_NAME, ZIP_CD, STATE_CD
001, CITY_A, 1111, AA
002, CITY_B, 2222, BB
003, CITY_C, 3333, CC

ijokarumawak

NiFi Example: How to Put/Get to/from a DistributedCache

Left part: Put

Set values you'd like to cache to the content of a FlowFile. Then pass it to PutDistributedCache.

Right part: Get

You can enrich incoming FlowFiles by adding an attribute with a value fetched from DistributedCache.

ijokarumawak

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<template>
    <description>This template generates messages, puts it to Kafka topic. Then another processor gets messages from
        Kafka and put it on HDFS.
    </description>
    <name>Kerberized Kafka and HDFS</name>
    <snippet>
        <connections>
            <id>2b93ffcd-0698-44a9-86f6-ce0ea6fc4145</id>
            <parentGroupId>3bdd324d-db87-4a21-8149-f88d7a46741e</parentGroupId>

ijokarumawak

This flow using Wait processor to wait for the all split FlowFiles to be processed. When it runs, following log messages can be seen:

2019-04-05 10:02:58,752 INFO [Timer-Driven Process Thread-7] o.a.nifi.processors.standard.LogMessage LogMessage[id=eafe4890-0169-1000-9927-a81fbbafafc6] Processing 1/5 message for 875bd765-209d-4136-a75d-5d14241384fe
2019-04-05 10:02:58,769 INFO [Timer-Driven Process Thread-3] o.a.nifi.processors.standard.LogMessage LogMessage[id=eafe4890-0169-1000-9927-a81fbbafafc6] Processing 2/5 message for 875bd765-209d-4136-a75d-5d14241384fe
2019-04-05 10:02:58,770 INFO [Timer-Driven Process Thread-3] o.a.nifi.processors.standard.LogMessage LogMessage[id=eafe4890-0169-1000-9927-a81fbbafafc6] Processing 3/5 message for 875bd765-209d-4136-a75d-5d14241384fe
2019-04-05 10:02:58,770 INFO [Timer-Driven Process Thread-3] o.a.nifi.processors.standard.LogMessage LogMessage[id=eafe4890-0169-1000-9927-a81fbbafafc6] Processing 4/5 message for 875bd765-209d-4136-a75d-5d14241384fe
2019-04-05 10:02:5