BYOP: Custom Processor Development with Apache NiFi

[Music] good afternoon everybody that's a little okay there we go my name is Andy LoPresto welcome to BYOP bring your own processor this is custom processor development in Apache knife I appreciate you being here I know it's getting towards the end of the conference everybody's tired everybody learned a lot of new stuff that they want to go home and use so hopefully we can just add a little bit more on to your conference experience and this is the first time that this presentation is being given anywhere so we're gonna learn it together there's a lot of information here all of these slides are going to be made available online as well as the audio recording so hopefully if there's something that you you miss or you want more detail on you can catch that again a second time but there is another session after this by Monica franchise Jeannie on knife I as a cybersecurity tool that I really want to go see so I got to get out of here at 4:40 this is a fairly advanced talk in the knife eye ecosystem so I'm not gonna spend much time on introduction to knife i covering generic topics and concepts there's a little bit of expected knowledge here if at the end you have questions about something please let me know so starting with knife I one-minute intro to the ecosystem we know that knife eye is a server data center class application it has the user interface it has the REST API it interacts with hundreds of services from out-of-the-box processors minify is our remote agent class service that's running on restricted Hardware limited hardware giving us lots of visibility out to the edge and data collection at the origin of data and then knife eye registry is this complimentary application that allows for flow versioning and in the future extension versioning this is the knife I user interface for anybody who hasn't seen this before you're probably not gonna like this talk but this is what we're working with so out of the box 286 give or take processors that come with IFI if you have maybe existing services you've already worked with or your company has some custom processors that you've interacted with that would be in addition to these generic Apache released processors so let's talk about extending the functionality of knife I with 300 processors what else is left to do right why do we need any new processors this is certainly not an exhaustive list but some of the use cases where building a custom processor makes a lot of sense is for interacting with a custom or a proprietary protocol that your company has internally or your organization has internally a great example that Neels you know and i were talking about earlier today was reading from a serial interface so if you're running on a Raspberry Pi interacting directly with a device that's on the GPIO pins or on the you know USB connection rather than having to use some other tool to translate that to standard input or a file transactional operations so we have lots of processors that interact with AWS s3 for example and a very common flow is to pull data off of a bucket do some transformation and then push it back on to another bucket well that works in knife I out of the box you make three processors you can you know fetch s3 objects convert record put s3 object but those steps are all happening independently and so each operation is transactional within the knife I system right you have consistency you have persistence in your content repository you have flow file repository so that you go back and replay those flow files if you wanted to you have Providence repositories storing all the metadata around those operations but it doesn't enforce a total transactional guarantee on the interaction with the AWS system right you might get into a state where you've pulled files off of the ATM the first s3 bucket you've done some transformation your knife I cluster crashed not because of knife I because of hardware problems and then it didn't get put back to the destination s3 bucket right so the entire operation wasn't completed in a transactional manner if you write a custom processor that does all three of those steps you could enforce that experimentation right just playing around with new features new opportunities and processor it's a great way to do that a commercial API or SDK that we don't maybe can't distribute with a knife I because it's not licensed for Apache distribution or then working with legacy code right a lot of times this is more of a customer request rather than a user request in the open-source community but working inside of for example financial institutions where there's software that was written 20-30 years ago which is verified and we know that it does the job it needs to do but we have no visibility into the code the person who wrote it left or was promoted or was demoted many years ago and no longer can manage this so we have a team of seven people whose job is just to make sure this Perl script keeps working right as far as custom development goes there are a number of different ways that we can achieve a lot of these goals the simplest one so along the bottom here I have effort or complexity the simplest one is execute process execute stream command those processors allow you to run basically any command that you would run at a terminal shell and read from standard input write to standard output and have the flow file contents for execute process captured for execute stream command both fed into the command and read from the output getting a little bit more complex but giving you a little bit more capability which is the size of these circles here would be execute script or X or invoke scripted processor with a special note on execute groovy script that it's groovy specific but it gives you a little bit more flexibility and we'll get into that later execute script invoke scripted processor allow you to choose from one of six scripting languages groovy Ruby Python really jython Lua closure and JSON excuse me JavaScript to write custom logic and run that inside of an immediate wrapper inside knife I you don't need to restart your knife I instance you don't need to deploy code to it all you're doing is reading from a local file on the file system or copy and pasting scripting code directly into the processor properties and having that run and we'll get into the details of that and then finally a custom sir the most complexity to build the most performance the most capabilities so some quick overviews of each of the processor types I just discussed execute stream command like I said pipes the flow file content in to standard in and populates the outgoing flow file content from standard out you can also direct that output to an attribute if you wanted to keep the flow file content that was there previously in this case I'm just piping content to the Rev command on the UNIX shell and so it's reversing the flow file content execute process is similar but there's no incoming flow file so you can use execute process to start a flow it doesn't require a flow file coming in to trigger it in fact it doesn't accept a flow file coming in to trigger it this is an execute script processor as I mentioned six scripting languages available anything that implements the jsr 223 spec there is a call-out for Python that it's actually JSON which in general is fine but you cannot use compiled C libraries in that code so that can be a challenge for people the way around this is to either use execute stream command to invoke Python on the shell and run your script that way or with a future upgrade of this processor that doesn't require the jsr or two to three there would be something like an execute Python script processor that does Python specifically and then handles compiled C libraries you can put as you see here the code directly into a property descriptor on this processor or you can reference it on a file on the file system there are pros and cons to both of those if you're referencing something on the file system you can have that file checked into version control you could have multiple people collaborating on it just as you would with any other normal software development environment the downside is that the content of that script is not captured in knife I flow versioning so if you copy and paste the script in here that means that when you version this flow in an eye-fi registry instance and somebody else brings it into their instance of knife I this content is kept this source code comes along with it so different capabilities of versioning given those two choices execute groovy script is a processor that was contributed by somebody in the community so not a core committer it's certainly supported but it allows for some groovy specific benefits because it makes the trade-off of not supporting any other scripting language so for example this code right here is functionally equivalent to no actually this is the JSON one this is reversing the text and you can see on that fourth line there that is using you know groovy method chaining the fifth line is writing directly to an attribute by using groovy syntax of just referencing a property on an object and then being able to transfer it directly to the success relationship this also allows for controller service access which regular execute script does not execute script wraps this and basically puts it into a non trigger method of a generic processor so within that method you have success and failure relationships you have a log you have a session context and process process session available to you but you don't have the controller services so if you want to do things with sequel for example and you have a JDBC or a DB C pecan connection service using the execute groovy script processor will let you access that directly there's a number of additional details on the documentation page for this processor it's fairly new it's it's pretty useful and then invoke scripted processor is a little bit more in-depth it's essentially taking the script that you would have put just in your on trigger method doing that more explicitly so you are constructing a processor in whatever scripting language you are using in this case I'm using groovy I'm building a groovy processor class and I can override the initialize method the relationships and property descriptors getters and I can put code directly into on trigger now this also allows me to compose my class a little bit more cleanly so I can refactor and have methods that I'm calling from on Twitter as opposed to a single you know inline script that's kind of spaghetti-like it also supports custom relationships so I can start building out additional relationships that I might want to send data to as opposed to being restricted to just success and failure this is also faster than execute script because execute script recompiles that script code every time on trigger runs and that's I'm sorry not on every time on trigger runs every time that on scheduled runs when that processor is started so that can be a fairly heavy overhead cost whereas invoke scripted processor does that once and then has that code ready to go every time but the reason you're all here is that we've exhausted our scripting capabilities and we want to actually build a custom processor right you want the experience of dragging a box onto the screen with your code already in it so custom processors do provide high performance because it's compiled code it's already available in the system they provide reusability and convenience and Virginie right I conversion these extensions just as I would any other processor that comes in knife I by default so the first step is to use the maven archetype to build our maven structure around this everybody's familiar with maven I'm assuming it's a build and dependency management tool for JVM so there is an archetype published for an Apache knife I naar which is simply a knife I archive right it's our module for containing all of the relevant files to a knife I processor component when you run this command you'll be prompted for group ID our artifact ID version those our maven standard values right so if you have comm dot your company or I don't know condom yes dot your company or yes contact your company that's gonna prompt you for those values and it'll build that out and populate the pom dot XML as it normally would you then have a maven project structure so you can see here it's just echoing all those values back to me saying build was successful echoing here build successful so once I have that maven structure populated this is what it looks like and so you can see that there's really four files of any relevance here there's a you know directory strings that get built out but outside of that hierarchy for files that I care about one is the my processor Java one is this org Apache knife I processor dot processor that is a manifest file which lists all of the implementing classes that can match this my processor so in this case I only have the one implementation and that's the single line that's in that but there are other if you look at like knife i standard bundle for example you'll see knife i standard processors which is something like sixty three and that's route text route on attribute route on content things like that those are all listed in that manifest file so that the process knife i framework knows which processors are present when it loads that nar finally a unit test is provided by default unfortunately it's empty you have to still implement the unit test we don't do magic for you and then the pom.xml which describes the structure of the module so this is the first of two slides on metadata around the processor so we've opened my processor java and this is after the license and import statements this is the first thing we're going to see each of these annotations has you know obviously a purpose not all of them are populated every time we encourage you to populate as many as you can and as make sense so tags when you do that ad processor dialogue and you have a tag cloud down on the bottom left tags or what will populate they so that you can filter very quickly to find your processor or relevant processors the capability description is what's shown underneath that it's a description of your processor so this is usually one or two lines it's not super detailed but it explains the purpose of your processor if you're writing a custom processor I'm assuming that you understand why you're doing that I hope you do but for other people who might be using your processor in the future this can seem like well I obviously know why I'm building this processor it's helpful for other people remember that this may be deployed for what we always say is proof of concepts end up in production you write this custom processor as you're fooling around and you're gonna find out 10 years later your company relies on it for production value see also simply lets you essentially link to other processors that might be related so for example on my put s3 object I would say see also fetch every object and reads writes attributes is really useful for explaining to people what they can expect for the incoming flow file requirements and the outgoing flow file expectations invoke HTTP excuse me invoke HTTP for example will have a very very long list of the attributes that it writes because it populates many of the HTTP response headers in as attributes so will enumerate those so that you know what to expect when you get a response back some other annotations which are not as common the ones that I just mentioned are all there by default when you generate a new custom processor these you have to add manually as relevant so event-driven would indicate that a timer won't control the scheduling of this processor this processor runs when it gets input right when it gets a flow file side-effect free fairly self-explanatory supports batching means that the operation that it's going to perform it could handle reading one flow file or reading a hundred flow files at once and it operates on them independently but it can operate on multiple ones in a single execution cycle so there are some processors where this is not supported and every time it reads from the queue it only reads one flow file operates on that completely isolated most processors are built in a way that they can take a group of flow files operate on each of them independently and persist them out to the resulting queues and connections but without having any they don't have to do a hundred execution cycles to do that input requirement means for example it can't be the first processor in a flow so if it was reverse incoming text for example you can't start a flow with that you need some flow file which has content to be coming in to that processor system resource consideration it doesn't do anything technically it doesn't tell the framework to do anything it's just for documentation to warn users that they need to be careful with what kind of data they're feeding in here things like XML parsing where it has to build the entire object model you don't want to feed a two terabyte XML file into the processor without knowing in advance that it's gonna try and build that and put it in memory right most of the processors are designed to be streaming so they're very memory safe but there are some where simply by a virtue of what the processor is doing it has to load some large data structure and so we want to make sure that we warn users before they start piping giant amounts of data into that experimental that's up to you whether you want to put that on there most of the processors that have that have that because they're not officially supported in production by open source Apache knife I so for example when Matt Burgess writes a new processor that does execute Python script the first few releases that'll probably be marked as experimental until we can gather enough data from real production use cases to see ok we're not encountering as many errors as we thought the performance is better than we expected things like that at restricted does have special technical meaning though at restricted will give you a little shield icon on the processor if you're in any kind of authorized environment so knife I with TLS enabled users need a special permission to be able to modify or create restricted processors and the reason for this is we have very fine-grained access control on all the search process groups things like that and that's really great restricted processors have an additional capability to potentially do damage to your system things like get file put file execute script if we're allowing you to run arbitrary code on your system we want to make sure that whoever setting up the authorization is explicitly aware that you can do that right I don't want to get a complaint that Oh a user ran get get file on Etsy password and I had no idea that they could do that right so there's certain processors I believe it's 19 or 20 at this point out of almost 300 that have that annotation so it's not a high number we expect that if you're using knife I you're generally trusted to do some kind of data transformation but things that can impact the system that knife is actually running on in a direct manner usually get that kind of annotation so now we implement the code right this is the easy part this is where you're all excellent software engineers I don't have to explain this very much we write code we put it into the on trigger method and this is the method that gets called every time knife I execute that processor we've set up our property descriptors and our relationships so for anybody not familiar a relationship is the connection that comes out of knife I write usually success or failure but there can be multiple for example route on attribute can have an arbitrary number of output relationships as you're matching to different patterns the property descriptors are when you right-click on a processor and say configure those key value pairs are property descriptors so it's a wrapper around some kind of property whether it's a string a number a boolean a controller service reference includes description includes validation and a list of allowable values if necessary also includes the option to mark a value as sensitive so if you mark a value as sensitive when the first value is put in it will be immediately encrypted persisted in the flow XML GZ in an encrypted format it will only be decrypted in memory to be used it'll never be returned via the API and plain text at all even if you're the user that said it it's also not exported as part of a template or saved in flow versioning in the version control system so what that means is when you deploy versioned flow to a new knife I instance or even the same instance again you'll have to manually populate sensitive properties now it's smart enough that you only have to do that one time so when you upgrade versions later it won't overwrite that with an empty value because it knows it was a sensitive property that you set when you set when you write a custom processor things like database passwords credentials for Amazon services etc you're gonna want to mark those as sensitive values so in this case I did add a custom stream callback class at the bottom there's last six lines of code all that's doing is implementing a standard interface and ëifí accepting an input stream in an output stream and then performing the the read/write logic in this case simply reversing the line of text that's there I reference that stream callback from the on trigger method again this was very quickly done for a presentation this is certainly not refactored or highly performant it's a hundred lines here because of all the template boiler code stuff that comes automatically but there's about thirty two lines that are needed to make a minimally viable processor so testing how are we doing on time ok write the test first right so in my normal development life cycle this section would be before writing the processor I like test-driven development and behavior driven development but understood that it's not everybody's cup of tea so we'll talk about test after we've written the processor we do have a test runner class which you can see I mean literally hundreds of thousands of examples in the existing knife I code this is how we unit test every processor that we release handles in queueing arbitrary data arbitrary flow file content attributes it allows you to make some intelligent assertions so if you look down here at the bottom of the test in this assert block here it's testing that specific flow files went to a relation ship you can get from that Q you can assert content equality attribute equality there's methods like assert attribute present assert actually not present so there's a lot of really helpful methods there that don't make you reinvent the wheel and you can see my favorite the green checkmark that all of my tests pass that's not a lot on testing and we'll come back to it what we need to do now is deploy our processor right so we build our maven project right we want maven clean install and that compiles the source code compiles the test runs the test and then outputs my target files so in this case I get a NAR output very similar to running maven clean and saw and gating a jar output in your target directory once I have that NAR I want to copy it into my knife I home directory and the Lib directory and under underneath that when I start knife I it'll automatically load this bundle and it'll parse out all the extensions from that manifest list and it will load each of those so here we have highlighted proof that it is loading that processor so when I go to my ad processor dialog I can now filter by my custom group right which was the package that I've put that in I can search by tag name by processor name and you can see the custom group the description here and the version of 1.0 snapshot so now it's time to configure that processor in this case I'm going to feed it with some data that I've put into a generate flow file processor so I have custom text I just printed the English alphabet and I'm gonna run it through reverse text and then log the output so we call this a smoke test right I just don't need to go and evaluate the temperature of every square foot I want to see if there's smoke coming out that tells me there's a fire in this case I'm listing the Q that's on the success relationship and I can see that the attribute reversed was set to true which is okay that's code that executed in my processor and I look at the content and I see that the alphabet is now backward so to me this is a passing test now I mentioned putting the processor into your knife I home slash Lib directory if you've worked with a knife I at all in the past you understand that's how NARS get detected and loaded in the recent releases of ninety five one nine zero one nine one now if I can now hot load NARS that are put into the extensions directory while this is useful for running in you know test or production mode where you get a new version of something and you want to load it in without restarting your instance or your cluster I wouldn't recommend this for developing custom processors because you're gonna be publishing the same file name every time right your your NAR is gonna have the same name so this system uses file pattern matching on the name to detect new nars to pick up now you could have it build script in your maven goals that appends the timestamp or something so that it will pick it up but you'll end up with how good you are how many versions of that processor you have to compile you know two to two hundred versions of the same processor in your add components dialog and that may or may not be comfortable for you so let's get back to testing in code there are a few approaches to doing testing you know this is not a evangelism session on testing I think that's obviously a good idea but that's up to you individually but there is unit testing integration testing and then testing within knife I so unit testing the benefits are you know very clear we support regression testing whoo it allows us to extract behavior from the processor class that we built initially into maybe an external service for reusability for componentization for just good design patterns and it allows us to test various combinations of properties and so as we can the processor with this I want this property true this property false this property be less than 10 I can run that against the same sets of content and attributes and ensure that I'm covering all my edge cases there are there are a number of mock classes but three that I want to call out in particular a mock flow file allows you to do internal assertions on the content and attributes of the flow file the mock process session allows you to do things to a flow file so simulate creation transfer cloning things like that and the mock process context allows you to interact with the properties of the processor controller services etc and all of this is done without having to read and write to the file system like a normal content repository flow file repository Providence repository would require you to do so unit testing for real saving you the interaction with the file system that you sometimes incur but wouldn't really want in this case I personally prefer groovy unit testing so all of the production code in knife I anything in source slash is written in Java because we value community understanding and contribution over particular language benefits right we want anybody who understands Java to be able to come in and understand well at least read what's happening throughout the entire system of Taif I understanding may take a little bit more time but if we start accepting Scala groovy JSON components and internal frameworks that really fractures the community and can make that much more challenging to support however for things that are not production code like tests we do allow for groovy tests for Spock tests for things like that I personally like groovy because of features like map coercion and meta class overrides it gives me the ability to write code that reads more linearly to me makes sense it's quote more expressive closer to English than code and I let the computer deal with translating that to JVM bytecode just as it would with Java so in this example I'm actually mocking a key provider in the encrypted provenance repository implementation where I get to have any arbitrary method called against the logger that I want because I've overridden the method missing and I just have an info that prints out what I was trying to do with that statement and the same time I have to mocked methods get key and key exists both are gonna return static values but I'm at least checking what's happening you can do this with a lot of mocking frameworks like mojito or power mock I like to do it in groovy because it's just a little simpler and there are fewer dependencies so then integration testing right we actually do want to test if this is going to interact with an external system well one of the really well I don't say really easy ways but one of the nice ways to do this is through docker containers so there are a number of integration tests in core knife I which use an external docker container that spun up as part of the test execution starts with a service like MongoDB or Kafka and then runs programmatic tests against that makes the assertions and then cleans up after itself so a good example of this is the MongoDB integration test suite these do not run by default when you build an eye-fi because it's already depending on your machine 2025 minutes so we run unit tests only but with a simple maven profile enablement you can run the integration tests as well sit back and order a pizza because it'll take a while and then we get the testing within knife I so knife I it's a wonderful tool it has some limitations but it also has some really good use cases that it enables one of those is building these very flexible and dynamic flows and that enables testing using the framework itself so one of the things one of the patterns that I follow quite a bit is creating a process group that has these test fixtures in it so that I can plug that into whatever flow I'm building at the time and as soon as I have the test data flowing through in a way that I like I can just stop that process group start whatever my production process group is or ingest data from wherever the actual sources so maybe I would normally be calling an HTTP endpoint from Twitter or Yahoo weather or a sports site something like that instead I can call that API once copy the JSON that I receive put that into a generate flow file or put it into a file on disk and run caddy on my local system and I can hit that endpoint as many times as I want without worrying about incurring you know network costs or hitting some kind of API threshold for the day or hitting quotas things like that I can execute as many times as I want against static data or dynamic data by using expression language to generate values inside of that and I can verify that it's working the way I expect and then I Stern off that process group connect to my real API in a way I go so some other capabilities this this slide came from a discussion I had at lunch with Niels sometimes having the normal property descriptors doesn't really make sense so in this case this is just a simple update attribute processor Niels has a log parser project that if it was configured as an eye-fi processor you'd have basically over a hundred boolean property descriptors here and they'd all be required so maybe they all default to false but you have to go through and scroll until you find the thirty-seventh and turn that to true and then the 85th and turn that to true that's not a great user experience here right so there is the capability to do custom user interfaces update attribute has one for example jolt transform JSON has one for example where you have a simple user interface like this but then down at the bottom I have an advanced button and if I click that advanced button I get an additional user interface which is designed per processor or per component and it's done it allows for custom user experience you could do really complex validation rules here you could have user feedback you could do like a wizard and walk through a couple steps of configuration so this is really kind of limitless here it allows you to really build the user experience that you want for that processor if it's too complex to do in a generic system the code for it is pretty simple the update attribute a little bit more complex because actually supports some rules so there are some dto and end to the objects here but in general you just need the one class and then a JSP and images if you want but CSS and JSP for presentation Java for objects in logic and you put this into a module a maven module which is just you know whatever your processor is - UI there's also a custom content viewer so if you've used knife I when you go and examine a flow file in a queue and you can see the attributes and you there's also a capability to view the content of that flow file in general we're looking at text or binary and that gets rendered in a standard content viewer I can see it as original which is just usually a skier utf-8 encoded I could see it as formatted so if it's something like JSON it'll do a printing formatting and you know nesting and indentation or I can look at it as hex where I can see the rendered bytes as well as the hex encoding line-by-line in line but that's that works for text that works pretty well for XML and CSV and JSON but what about other things what about other media types so an image is supported by default and you can see here it's detected that the content type is a PNG image so I chopped chopped it off but it was a screenshot of knife I itself but something like a video there's no viewer registered for this content type you can do that though you can build a custom viewer and register that so I could have inline video playback here or I might have a Visio file or some kind of diagram that's not a standard image format that I've registered a custom content viewer and that lets me look at it inside of the knife i interface so I think for some people if you're in an organization where you have these custom file types and formats that might be some gears turning now of oh we could import that into knife I and be able to visualize our custom data format inside of the tool again it's just a simple Java class JSP and CSS for presentation so let's talk some best practices these are no longer requirements for your thing to work these are recommendations using execute script and evoke invokes scripted processor execute script is easier for rapid development because you just change one line of script copy/paste you're good you can even edit in the process the processor property if you want but invoke scripted processor is definitely more performant if you're doing this at any kind of production flow where you have a custom custom script it should always be in the ISP not in execute script it also allows for custom relationships so it's a little bit more flexible module organization usually it's it's fine to co-locate your processor and a simple controller service implementation but if you're making controller services that have an interface and you expect there to be multiple implementations or you expect somebody else to extend that it's really nice to put those in separate modules so they don't need to depend on all of your code in order to implement that module configuration check the archetype version check your knife I a P I version when you're building do your initialization of relationships and property descriptors in your init method not in your on trigger where it would have to build every time that the processor runs on the other hand don't connect to external services in on scheduled or NIT do that inside of on trigger because otherwise you might retain a connection that's not used for a long time and it times out and now when on trigger runs you're left you thought you had a connection but you don't you can handle dynamic properties you can do custom validation there are a number of validators in the standard validators class that give you a really good example for how to build those good documentation handle flow file content and streaming manner generate your proper provenance events and define appropriate relationships like these are all just good things to do make your processor easy to use really going tight on time so i'm going to fly through here always put your processor in the manifest list this is the number one stupid bug that i our stupid mistake make when i'm building a custom processor and then i deploy it and it doesn't show up i go back and look at the manifest list i forgot to put that one line in there so this one is one that i think we'll go back and look at online I'm not gonna go through all of it basically performance trade-offs complexity trade-offs usability trade-offs that's that's what this slide is everybody got a picture before I go on okay performance testing generate flow file I filled all of these queues by using back pressure and then I'm just comparing us using standard in two UNIX reverse standard out a custom processor and a groovy script spoiler alert the execute process not very good so it's not even on the next slides performance results this is a very simple task this is not optimized this is not you know was not refined at all but the custom processor is faster than using execute script okay more performance results notes on performance additional resources there's links to some great guides on this new announcements C++ is being voted on right now if you like C++ go vote on that one and more talks go to Monica's taco to Tim's talk those are the last two sessions of the day thank you very much [Applause] you

Channel: DataWorks Summit

Views: 5,904

Keywords: IoT and Streaming Analytics, dataworks summit barcelona, dws19, apache nifi

Id: v2u0WsPs2Ac

Length: 41min 28sec (2488 seconds)

Published: Tue Apr 23 2019