Hi there, I have something I am a bit stuck on. Within my flow my avro.schema changes because I have the flow files going through a few custom processors to change the data and add new fields to the data. At the point in which the data is coming out of the custom processors, its in json format, so I can't re-use the same "extract avro metadata" processor. I found that I could use the "infer avro schema" processor and just have it set the attribute "inferred.avro.schema" with what the new schema should be. At that point I can update the avro.schema attribute to be that of the inferred... but I found a few issues with the formatting of the "inferred.avro.schema" processor. It seems to be very verbose and hive won't accept that output as part of a create table statement. Because of this I have started to regex out the pieces that I know are not formatted correctly. I have most working, but stuck on one item. I don't see where I can start and end a string literal. Example dummy output snippet, from inferred.avro.schema: {"type":"record","name":"test_record","fields":[{"name":"bill_id","type":"string", '\"488\"'"},{"name":"congr_district","type":"string", '\"YEA\"'"},{"name":"cq","type":"string", '\"2004\"'"},{"name":"day","type":"string", '\"On Motion to Suspend the Rules and Pass\"'"},{"name":"month","type":"string", '\"4\"'"},{"name":"pull_date","type":"string", '\"2018-02-26\"'"},{"name":"pull_date_normalized","type":"string", '\"2018-01-26 00:02:00.000+0000\"'"},{"name":"question","type":"string", '\"nan\"'"},{"name":"rep_name","type":"string", '\"DEMOCRAT\"'"},{"name":"rep_party","type":"string", '\"OHIO\"'"},{"name":"state","type":"string", '\"24\"'"},{"name":"state_code","type":"string", '\"17\"'"},{"name":"title","type":"string", '\"encrypted_content"'"},{"name":"vote","type":"string", '\"To transfer Federal lands between the Secretary of Agriculture and the Secretary of the Interior\"'"},{"name":"year","type":"string", '\"OCTOBER\"'"}]}} So you will notice that I have parsed out the "Schema inferred by...." stuff, but I need to remove the items within the \. So this: {"name":"cq","type":"string", '\"2004\"'"},{"name":"day","type":"string", '\"On Motion to Suspend the Rules and Pass\"'"} Would be this: {"name":"cq","type":"string"},{"name":"day","type":"string"} I can't find a good way to start and stop within the expression language. I know for sure that regular java regex would pick it up as this: '\\.*?} Where it would start at the first \ and then go until it finds the first }. I don't see how it can be achieved from Nifi EL. It will accept this format, but don't do anything to the output of the attribute (it remains unchanged). Anyone have any ideas? Thanks, Chris L ... View more

Hi all, I am in the process of migrating our current data pipeline into Hadoop from traditional tools (SSIS, Python, etc) to using Nifi. We have several hundred tables that run daily. I like the approach that Kylo is taking and while it doesn't satisfy my needs 100%, I have used a few ideas from looking at their general data templates. I am currently at a stage now where I am not entirely sure how to: (1) best optimize this process to run concurrently for ~50 tables at a time and (2) get the MergeContent to work correctly for how I am using it. I am still fairly new to Nifi, so I am looking for someone to help me with best practices, if possible. Quick background on the how the entire flow works: Step 1 -- I have one processor group for each table we are bringing in. Its categorized by our servers, database, then table. I was not able to find a great way to create a reusable flow for these as the controller service for the DB couldn't be dynamically set. Nonetheless, I have a bit of SQL executing to fetch the delta from the source. That then feeds into an InferSchema processor so I can grab the schema and have that as an attribute. Next the flow goes into an UpdateAttribute processor where I have it pick up some table specific attributes I have set, such as schema_name, table_name, hdfs_parent_location, etc... All of these attributes are there to help the flow be unique and I also use them to build the hive "create if not exists" and "msck repair" scripts at the end of the process. Step 2 -- Once that bit of data is set for each source, they all flow into one reusable template, using the Input and Output processors. This is where I am a bit concerned that I am doing things correctly. The reusable flow consists of the following: - AvroToJson - Custom Groovy Script to encrypt all the incoming PII (being read from an attribute the user would set) - Split JSON - JSONtoAvro: Sending the flow back to Avro - MergeContent: Merging all with Bin-Packing, Avro Format, Using ${avro_schema}_${tablename} as the Correlation Attribute and I have it set now to try and merge the files to around 64MB, using 5 bins. - PutHDFS: Dynamically creates the fully qualified location from the attributes and partitions data by "DAY". - Replace Text: I build the create statement and msck repair table to add any partitions. - PutHQL: Run statement Here are my questions: 1. I am not sure the MergeContent processor is working correctly 100% of the time. I have found that if I use larger tables (1-2MM records) at the same time, it generally fails at the MergeContent step. It seems that it has a hard time using the correlation attribute that I have set and I have a bit of trouble getting the file sizes to be around that 64MB size. For the correlation attribute issue, I find that the processors fails off records (I have failure route back to itself so I can check the queue) and they generally sit in the queue. I can see which table its coming from, but when I check the failure queue, all of the attributes which were previously set are now gone. I have checked them step by step and they don't get erased until the MergeContent runs. So that is a bit strange. For file size, my assumption is the bins are getting full, but I have only two tables running durning my test and the MergeContent processor is set to 64MB on min group size and 5 minutes for the wait time. So please tell me if I am missing something (I am sure I am). Also please let me know if you think the correlation attribute problem I am having is typical in how I am using the processor or if you think it should work. In production, it will be much more than 2 tables running at one time... probably 50 or so. 2. From what I have listed above does anyone think the flow for each table should be its own end to end or am I on the right track with having this single reusable template for which all flows go into? I would assume the thread count would be much less for the reusable approach, but do worry a bit about moving that much data through a single group. Any help in direction would be much appreciated!! Thanks, Chris @Matt Burgess @Matt Clarke ... View more