NIFI-4906 Add GetHDFSFileInfo #2639
Conversation
| .description("Regex. Only directories whose names match the given regular expression will be picked up. If not provided, any filter would be apply (performance considerations).") | ||
| .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) | ||
| .required(false) | ||
| .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR) |
There was a problem hiding this comment.
Should this use a regex validator like the following?
StandardValidators.createRegexValidator(0, Integer.MAX_VALUE, true)
The true means it would only validate the regex when non-EL was present.
| .description("Regex. Only files whose names match the given regular expression will be picked up. If not provided, any filter would be apply (performance considerations).") | ||
| .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) | ||
| .required(false) | ||
| .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR) |
| .description("Regex. Files whose names match the given regular expression will not be picked up. If not provided, any filter won't be apply (performance considerations).") | ||
| .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) | ||
| .required(false) | ||
| .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR) |
| import org.apache.nifi.processors.hadoop.GetHDFSFileInfo.HDFSFileInfoRequest.Groupping; | ||
|
|
||
| @TriggerSerially | ||
| @TriggerWhenEmpty |
There was a problem hiding this comment.
@TriggerWhenEmpty is used to make the framework trigger the processor when there is an empty incoming connection, is that necessary in this case?
You typically use this if there is a case where you need to perform some kind of check/clean-up even when no flow files are available.
There was a problem hiding this comment.
@bbende , removed TriggerWhenEmpty and TriggerSerially. Copy/paste from statefull processor, no need in this case.
| }) | ||
| @SeeAlso({ListHDFS.class, GetHDFS.class, FetchHDFS.class, PutHDFS.class}) | ||
| public class GetHDFSFileInfo extends AbstractHadoopProcessor { | ||
| static final long LISTING_LAG_NANOS = TimeUnit.MILLISECONDS.toNanos(100L); |
There was a problem hiding this comment.
Nitpick, but this is unused
| context.yield(); | ||
| }else { | ||
| ff = session.penalize(ff); | ||
| session.rollback(); |
There was a problem hiding this comment.
If we couldn't build a request for the current flow file because some properties were incorrect, should this be routed to failure instead? or maybe an "invalid" relationship to indicate a difference from other failures?
By rolling back we are putting the flow file back in the incoming queue (if we got it from the queue) and I would expect it would keep failing over and over if it didn't work the first time.
There was a problem hiding this comment.
at some point I managed to handle build and update without throwing exceptions at all, but forgot to remove try/catch. So removing it now. good catch.
| /* | ||
| * Since, by definition, FF will keep only attributes for parent/single object, we don't need to recurse the children | ||
| */ | ||
| public Map<String, String> toAttributesMap(){ |
There was a problem hiding this comment.
If the destination is content, then we should be setting the mime.type attribute to application/json
There was a problem hiding this comment.
toAttributesMap isn't used when destination is content (toJsonString is being used instead). But I do agree on this miss - going to add it under processHDFSObject when I create new FF.
| session.transfer(ff, REL_NOT_FOUND); | ||
| return; | ||
| } | ||
| session.transfer(ff, REL_ORIGINAL); |
There was a problem hiding this comment.
In the case where there are no incoming connections, we end up creating a new flow file which ends up getting transferred to original. Should we wrap the transfer to original in a "if !scheduleFF" so that it only happens when there was an incoming flow file?
| UserGroupInformation ugi = getUserGroupInformation(); | ||
| HDFSObjectInfoDetails res = walkHDFSTree(context, session, ff, hdfs, ugi, req, null, false); | ||
| if (res == null) { | ||
| session.transfer(ff, REL_NOT_FOUND); |
There was a problem hiding this comment.
In the case where there are no incoming connections, the "ff" flow file has no info about what was not found, should we add an attribute with the path that it was looking at?
There was a problem hiding this comment.
@bbende , good catch. although, in such case there will be properties of the processor used to compute path, and other params, it would still be nice to have computed values at run time. adding this info now.
| + "3 for the group, and 3 for other users. For example rw-rw-r--"), | ||
| @WritesAttribute(attribute="hdfs.status", description="The status contains comma separated list of file/dir paths, which couldn't be listed/accessed. " | ||
| + "Status won't be set if no errors occured."), | ||
| @WritesAttribute(attribute="hdfs.full.tree", description="When destination is 'attribute', will be populated with full tree of HDFS directory in JSON format.") |
There was a problem hiding this comment.
I'm wondering if we should only allow destination of attributes when group is set to none, because otherwise this could create quite a large string for a flow file attribute. Imagine the size of the JSON document if you had grouping "all" with recursive "true" and found 10k files/dirs. Large flow file attributes can negatively impact the flow file repo as well as GC/heap usage, but maybe it is a use-at-your-own-risk situation?
There was a problem hiding this comment.
@bbende , yes, I was thinking about this use case too. IMO we shouldn't limit, but I'll add warning to both property of destination, and to the output attribute description.
Changes: - changed validators for regex params (3) - removed TriggerWhenEmpty and TriggerSerially. - removed unused LISTING_LAG_NANOS - removed unimplemented customValidate - removed unsused error handling for building/updating request params - added MIME_TYPE for content - suppressed output to original relationship when scheduled - added full path to original/new FF when not found - added warning messages for huge attribute values.
|
@bbende , thank you for your comments! I've made the changes to address all your concerns. Please review once you have a time. |
|
Looks good, will merge, thanks! |
NIFI-4906: Added support to scan HDFS to get files and directories information without pulling the files.
Thank you for submitting a contribution to Apache NiFi.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
For all changes:
Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
Does your PR title start with NIFI-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
Has your PR been rebased against the latest commit within the target branch (typically master)?
Is your initial contribution a single, squashed commit?
For code changes:
For documentation related changes:
Note:
Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible.