Introduction
Hortonworks Dataflow (HDF) powered by Apache NiFi, kafka and Storm, collects, curates, analyzes and delivers real-time data from the IoAT to data stores both on-premises and in the cloud. Apache NiFi automates and manages the flow of information between systems. NiFi data flows are made of series of processors each with specific task. NiFi provides hundreds of general purpose processors. NiFi can pull data from various sources. This document mainly discusses in detail the integration with AWS S3 data source.
Amazon S3 is cloud storage for the Internet. To upload your data (photos, videos, documents etc.), you first create a bucket in one of the AWS regions. You can then upload any number of objects to the bucket. S3 buckets and objects are resources, and Amazon S3 provides APIs for you to manage them.
With many existing AWS customers there is need to integrate with S3 to process data across multiple applications. NiFi provides many processors to manage and process S3 objects integrating with S3 buckets. This document outlines the detail setup and configuration to integrate S3 with Apache NiFi.
Business Use cases
Many customers are utilizing the Amazon S3 storage service to build applications.The application types can be backup & archiving, content storage, big data analytics, cloud native application data or DR.
S3 can be utilized as persistent or temporary storage. Many applications need to process the data as lands into a S3 bucket, retrieve the content and log the metadata regarding the S3 object. AWS supports following destinations where it can publish S3 related events.
- Amazon SNS topic
- Amazon SQS queue
- AWS lambda
This document describes putting and extracting data object from amazon S3 using Apache NiFi leveraging the Amazon SQS notifications.
Solution Architecture with Apache NiFi
The main purpose of the document is to showcase the ease of integration with S3 using Apache NiFi.
In the sample demo scenario:
- The cloud NiFi instance creates the data object to S3 bucket using PutS3Object processor.
- As the new object is created at S3, it sends out notification in JSON object form to amazon SQS queue.
- GetSQS processor subscribes to the SQS event and retrieves the newly created S3 object metadata.
- The FetchS3Object extracts the content of newly created object and sends it downstream systems for further processing.
AWS Configurations
This section describes the setup and configuration on AWS side (SQS & S3 bucket) based on the scenario described in the previous section. Make sure to login to the AWS dashboard and select appropriate product section to configure the S3 buckets and SQS queues.
SQS
One of the ways to monitor S3 bucket is to use SQS notifications.
First create the SQS Queue as shown below e.g. NiFiEvent.
Configure the security and appropriate permissions to the SQS queue so that SQS queue can be utilized for the S3 bucket events.
S3 Bucket
Once the SQS configuration is done, create the S3 bucket (e.g. mphdf).
Adding a folder named "orderEvent" to the S3 bucket. Go to the properties section and make sure to configure Permissions, Event notification and policy to the S3 bucket.
For permissions, add the appropriate account to include list, upload, delete, view and Edit permissions to the bucket.
Creating an AWS bucket policy, so that specific accounts have permissions to manage the S3 bucket objects.
The policy output will be JSON object which you can update in the bucket configuration.
Next configure the event notification to publish all the relevant events to SQS queues which we created earlier. Make sure to select the Type as SQS.
The AWS side configuration is now complete. Next, build the NiFi dataflow using the NiFi processors for S3 buckets and SQS.
NiFi DataFlow Configuration
To demonstrate the S3 integration I modified the existing NiFi data flow. There are two NiFi dataflows, one to publish the object to S3 and second one is to extract the object content through SQS notification.
Flow 1 : Accept the ERP Events --> transform to JSON --> PutS3Object
Flow 2 : GetSQS notification -->Get Object metadata --> FetchS3Object
The complete flow is shown below.
PutS3Object Processor
This processor creates the object in S3 bucket. The configuration is shown below. Make sure to use the correct bucket name, owner (aws account) , access key and secret key. To get the Access and secret keys go the AWS IAM console -> users ->security credentials->create access keys. You can download or select Show User Security Credentials options to get Access key and Secret access key. Make sure to select the correct Region.
GetSQS Processor
IT fetches messages from an AWS SQS queue. The Queue URL you can get from AWS SQS details section.
FetchS3Object Processor
Retrieves the contents of an S3 Object and writes it to the content of a FlowFile. The "object key" is nothing but the key attribute value from incoming SQS notification JSON message. The "objectname" property is populated by JsonPath expression $.Records[*].s3.object.key
Testing NiFi DataFlow
For testing purpose, the existing dataflow is used which:
- pulls the ERP events from JMS queue. The event xml is mapped to JSON object and the object is pushed to AWS S3 bucket.
- The successful creation of S3 object triggers a notification to SQS queue.
- The second NiFi dataflow reads the SQS event through GetSQS processor. It parses the JSON metadata information and extracts the object name from SQS message.
- Next it uses the fetchS3Object NiFi processor to extract the S3 object content and then writes it to the local file system.
Following are some data provenance screenshots from NiFi which proves the successful creation of S3 object and then processing the SQS notification to extract and process the S3 object through NiFi in near real time fashion.
PutS3Object data provenance
S3 Bucket browser view (shows the newly created object).
GetSQS data provenance
Showed the SQS event getting triggered for newly created S3 object.
Sample SQS Notification message
The highlighted value shows the object name which will be extracted and used to retrieve the object from S3 bucket.
{"Records":[{"eventVersion":"2.0","eventSource":"aws:s3","awsRegion":"us-east-1","eventTime":"2016-08-03T14:25:13.147Z","eventName":"ObjectCreated:Put","userIdentity":{"principalId":"AWS:AIDAJL3JQI6HZAG3MB6JM"},"requestParameters":{"sourceIPAddress":"71.127.248.137"},"responseElements":{"x-amz-request-id":"9A673206F1BFDE85","x-amz-id-2":"UmcqEKQJyXfH+UlgDTWIMfvQDOhuOWREe/lwUSJdMx9CbgCu7wzPWJL+wCeRzL6dgqsnYTopWrM="},"s3":{"s3SchemaVersion":"1.0","configurationId":"NiFiEvents","bucket":{"name":"mphdf","ownerIdentity":{"principalId":"A1O76DMOXCPR44"},"arn":"arn:aws:s3:::mphdf"},"object":{"key":"orderEvent/651155932749796","size":804,"eTag":"53ca61e19b3223763f1b36ed0e9383fa","sequencer":"0057A1FEC9137FBF3A"}}}]}
FetchS3Bucket data provenance
The content from the newly created S3 object extracted using the FetchS3Bucket processor.
Apache NiFi Benefits
NiFi is data source and destination-agnostic. it can move data from any data platform to any data platform. The S3 example demonstrates it can integrate well with non Hadoop echo system as well.
The NiFi project provides connection processors for many standard data sources like S3. With standard interfaces new processors can be developed with minimal effort.
You can create NiFi dataflow templates to accelerate development.
Apache NiFi is ideal for data sources sitting out on the edge in the cloud or on-prem.
Document References
NiFi System Admin Guide:
https://aws.amazon.com/s3/?nc2=h_l3_sc
https://aws.amazon.com/sqs/?nc2=h_m1
https://nifi.apache.org/docs/nifi-docs/
