If you have ever tried to spawn multiple cloudbreak shells you may have run into an error. That is because the default "cbd util cloudbreak-shell" uses docker containers. The fastest work around of this is to use the Jars directly. These Jars can be remotely run from your personal machine or run on the cloudbreak machine itself. Prepping the cloudbreak machine(only needed if running jars locally on the AWS image) Log into your cloudbreak instance and go to /etc/yum.repos.d Remove the Centos-Base.repo file (this is a redhat machine and this can cause conflicts) Install java-8 (yum install java-1.8.0*) Change directory back to /home/cloudbreak Downloading the Jar Set a global variable equal to your cloudbreak version (export CB_SHELL_VERSION=1.6.1) Download the jar (curl -o cloudbreak-shell.jar https://s3-eu-west-1.amazonaws.com/maven.sequenceiq.com/releases/com/sequenceiq/cloudbreak-shell/$CB_SHELL_VERSION/cloudbreak-shell-$CB_SHELL_VERSION.jar) Using the Jar Interactive mode (java -jar ./cloudbreak-shell.jar --cloudbreak.address=https://<your-public-hostname> --sequenceiq.user=admin@example.com --sequenceiq.password=cloudbreak --cert.validation=false) Using a command file (java -jar ./cloudbreak-shell.jar --cloudbreak.address=https://<your-public-hostname> --sequenceiq.user=admin@example.com --sequenceiq.password=cloudbreak --cert.validation=false --cmdfile=<your-FILE>) ... View more

In this article I will review the steps required to enrich and filter logs. It is assumed that the logs are landing one at a time as a stream into the nifi cluster. The steps involved Extract Attributes - IP and Action Extract Attributes - IP and Action Cold Store non ip logs GeoEnrich the IP address Cold store local IP addresses Route the remaining logs based on threat level Store the low threat logs in HDFS Place high threat logs into an external table Extract IP Address and Action - ExtractText Processor This processor will evaluate each log and parse the information into attributes. To create a new attribute add a property and give it a name(soon to be attribute name) and a java-style regex command. As the processor runs it will evaluate the regex and create an attribute with the result. If there is no match it will be sent to the 'unmatched' result which is a simple way of filtering out different logs. GeoEnrichIP - GeoEnrichIP Processor This processor takes the ipaddr attribute generated in the previous step and compares it to a geo-database('mmdb'). I am using the GeoLite - City Database found here Route on Threat - RouteOnAttribute Processor This processor takes the IsDenied attribute from the previous step and tests to see if it is there. This will only exist if the "Extract IP Address" Processor found "iptables denied" in the log. It is then routed to a connectionw ith that property's name. More properties can be added with thier own rules following the nifi expression language Note I plan on adding location filtering but did not want to obscure the demo in too many steps. Cold and Medium Storage - Processor Groups These two processor groups are very similar in function. Eventually they could be combined into one shared group using attributes for rules but for now they are separate. Merge Content - This processor takes each individual line and combines them into a larger aggregated file. This helps avoid the too many small files problem that arises in large clusters Compress Content - Simply saves disk space by compressing them Set Filename As Timestamp - UpdateAttribute Processor - This takes each aggragate and sets the attribute 'filename' to the current time. This will allow us to sort the aggregates by when they were written for later review PutHDFS Processor - Takes the aggregate and saves it to HDFS High Threat - Processor Group In order to be read by a hive external table we need to convert the data to a JSON format and save it to the correct directory. Rename Attributes - UpdateAttribute Processor - This renames the fields to match the hive field format Put Into JSON - AttributesToJSON - Takes the renamed fields and saves them in a JSON string that the hive SerDe can read natively Set Filename As Timestamp - UpdateAttribute Processor - Once again this sets the filename to the timestamp. This may be better served as systemname + timestamp moving forward PutHDFS - Stores the data to the hive external file location Hive Table Query Using the ambari hive view I am able to now query my logs and use sql-style queries to get results CREATE TABLE `securitylogs`( `ctime` varchar(255) COMMENT 'from deserializer', `country` varchar(255) COMMENT 'from deserializer', `city` varchar(255) COMMENT 'from deserializer', `ipaddr` varchar(255) COMMENT 'from deserializer', `fullbody` varchar(5000) COMMENT 'from deserializer') ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe' STORED AS INPUTFORMAT 'org.apache.hadoop.mapred.TextInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat' LOCATION 'hdfs://sandbox.hortonworks.com:8020/user/nifi/High_Threat' ... View more

I was recently tinkering with the walmart rest-api. This is publicly available interface and can be used for a quick price look up for products. The overall goal of the project is to keep track of the cost of specific shopping carts day to day but this intermediate step provides an interesting example-case. The requirements of this stage: Use UPC codes to provide a lookup Avoid having the pass the walmart API key to the internal client making the call Extract features such as price in preparation for a JDBC database entry. Core Concepts in Nifi Nifi has the ability to serve as a custom restful api which is managed with the "HandleHttpRequest" and "HandleHttpResponce" processors. This can be a Get/Post or any of the other common types Nifi can make calls to an external rest API via the "InvokeHTTP" Processor XML Data can be extracted with the "EvaluateXPath" Processor The HandleHttpRequest Processor This processor receives the incoming rest call and makes a flow file with information pertaining to the headers. As you can see in the image below it is listening on port 9091 and only responding to the path '/lookup'. Additionally the flow file it created has an attribute value for all of the headers it received, particularly "upc_code" And the flow file The InvokeHTTP Processor This processor takes the header and makes a call to the walmart API directly. As you can see i am using the attribute for upc_code received from the request handler. This then sends an XML file in the body of the flow file to the next stage The EvaluateXPath Processor In this article I covered how the xpath processor works in more detail. I am extracting key attributes for later analysis. https://community.hortonworks.com/articles/25720/parsing-xml-logs-with-nifi-part-1-of-3.html HandleHTTPResponse (Code 200 or Dead Letter) After successfully extracting the attributes I send a response code of 200(success) back to the rest client along with the xml that walmart provided. In my above example if i do not successfully extract the values the message goes to a dead letter que. This is not ideal and in a production setting I would send the appropriate HTTP error code. Closing Thoughts This process group provides a solid basis for my pricing engine. I still need to write in the error handling but this start provides a feature-rich flow file to the next stage of my project. ... View more

Recently I had a client ask about how would we go about connecting a windows share to Nifi to HDFS, or if it was even possible. This is how you build a working proof of concept to demo the capabilities! You will need two Servers or Virtual machines. One for windows, one for Hadoop + Nifi. I personally elected to use these two The Sandbox http://hortonworks.com/products/hortonworks-sandbox/ A windows VM running win 7 https://developer.microsoft.com/en-us/microsoft-edge/tools/vms/linux/ You then need to install nifi on the sandbox, I find this repo to be the easiest to follow. https://github.com/abajwa-hw/ambari-nifi-service Be sure the servers can talk to each other directly, I personally used a bridged network connection in virtual box and looked up the IPs on my router's control panel. Next you need to setup a windows share of some format. This can be combined with active directory but I personally just enabled guest accounts and made an account called Nifi_Test. These instructions were the basis of creating a windows share http://emby.media/community/index.php?/topic/703-how-to-make-unc-folder-shares/ Keep in mind network user permissions may get funky and the example above will enforce a read only permission unless you do additional work. Now you have mount the share into the hadoop machine using CIFs+Samba. The instructions I followed are here http://blog.zwiegnet.com/linux-server/mounting-windows-share-on-centos/ Finally we are able to setup nifi to read the mounted drive and post it to HDFS. The GetFile processor retrieves the files while the PutHDFS stores it. To configure HDFS for the incoming data I ran the following commands on the sandbox: "su HDFS" ; “Hadoop dfs -mkdir /user/nifi” ; “Hadoop dfs -chmod 777 /user/nifi” I elected to keep the source file for troubleshooting purposes so that every time the processor ran it would just stream the data in. GetFile Configuration The PutHDFS Configuration for sandbox And finally run it and confirm it lands in HDFS! ... View more

I have a plan to write a 3 part “intro” series as to how to handle your XML files. The subjects will be: Basic XML and Feature Extraction via Text Managment, Splitting and Xpath Interactive Text Handling with XQuery and Regex in relation to XMLs XML schema validation and transformations XML data is read into the flowfile contents when the file lands in nifi. As long as it is a valid XML format the 5 dedicated XML processors can be applied to it for management and feature extraction. Commonly a user will want to get this XML data into a database which will require us to do a feature extraction and convert to a new format such as JSON or AVRO. The simplest of the XML processors is the “SplitXml” processor. This simply takes the current selection of data and breaks the children off into their own files. The depth of the split in relation to the root is configurable as shown below. An example of when this may be helpful is when you have a list of events, each of which should be treated seperatly XPath is is a syntax language way of extracting information from an XML. It allows you to search for nodes based on hierarchy, name, or even attribute. It has limited regex integration and has framework for moderately complex queries. More complete documentation can be found here http://www.w3schools.com/xsl/xpath_syntax.asp The processor below shows the “EvaluateXPath” processor being combined with XPath language to extract node data and an attribute. It should not be confused for XQuery which I will cover in my next article. With executing the Xpath module something very important happens, the xml attributes are now NIFI attributes. This allows us to apply routing and other intelligence that is Nifi's signature. One of the transformations I have previously worked on is how to get the XML data into an AVRO format for easy ingestion. At this time all of the AVRO processors in nifi play nicely with JSONs so the “AttributestoJSON” processor can be used to as an out of the box intermediary to get the format you need. Note that I have set the destination of the processor to “flowfile-contents” which will over-ride the existing XML contents for a JSON. With a JSON + attributes this is a very easy flow file to work with and can be easily merged into existing workflows or written out to a file for the Hive SerDe. ... View more