diff --git a/nifi-docs/src/main/asciidoc/python-developer-guide.adoc b/nifi-docs/src/main/asciidoc/python-developer-guide.adoc index 7cc401a35bc4..cfec796cce74 100644 --- a/nifi-docs/src/main/asciidoc/python-developer-guide.adoc +++ b/nifi-docs/src/main/asciidoc/python-developer-guide.adoc @@ -122,7 +122,7 @@ Processors such as ExecuteScript) tend to be around data manipulation and/or com - Calls from Python to Java (and vice versa) are far more expensive than native method calls. Having APIs that are more tailored toward specific use cases allows for fewer interactions between the two processes, which greatly improves performance. -As a result, the Python API consists of three different Processor classes that can be implemented: `FlowFileTransform`, `RecordTransform` and `FlowFileSource`. +As a result, the Python API consists of five different Processor classes that can be implemented: `FlowFileTransform`, `FlowFileTransformMultipleOutput`, `RecordTransform`, `FlowFileSource`, and `MultipleFlowFileSource`. Others may emerge in the future. @@ -170,6 +170,51 @@ as it will have the same effect as passing `None` but will be more expensive, as Likewise, it is more efficient to omit the `attributes` unless there is any attribute to add. +[[flowfile-transform-multiple-output]] +=== FlowFileTransformMultipleOutput + +The `FlowFileTransformMultipleOutput` API provides a mechanism for routing and transforming a singl FlowFile based on its attributes as well as its +textual or binary contents into multiple FlowFiles, each with its own relationship and content. Contrast this with the `RecordTransform` API, which provides a mechanism for routing and transforming +individual Records (such as JSON, Avro or CSV Records, for example). + +In order to implement the `FlowFileTransformMultipleOutput` API, a Python class must extend from the `nifiapi.FlowFileTransformMultipleOutput` class +and implement the `transform(ProcessContext, InputFlowFile)` method, which returns an iterable of `FlowFileTransformResult` objects. + +Additionally, the Processor class must provide two pieces of information as subclasses: the Java interface that it implements +(which will always be `org.apache.nifi.python.processor.FlowFileTransformMultipleOutput`) and any details about the Processor, such as the +version, a description, keywords/tags that might be associated with the Processor, etc. +These will be discussed in more details below, in the <> section. + +As such, a simple implementation may look like this: +---- +from nifiapi.flowfiletransform import FlowFileTransformMultipleOutput, FlowFileTransformResult + +class WriteHelloWorld(FlowFileTransform): + class Java: + implements = ['org.apache.nifi.python.processor.FlowFileTransform'] + class ProcessorDetails: + version = '0.0.1-SNAPSHOT' + + def __init__(self, **kwargs): + pass + + def transform(self, context, flowfile): + results = [] + results.append(FlowFileTransformResult(relationship = "success", contents = "Hello from Flowfile 1!", attributes = {"greeting": "hello"})) + results.append(FlowFileTransformResult(relationship = "success", contents = "Goodbye from Flowfile 2!", attributes = {"greeting": "goodbye"})) + + return results +---- + +The `transform` method is expected to take two arguments: the context (of type `nifiapi.properties.ProcessContext`) and +the flowfile (of type `InputFlowFile`). + +The return type is an iterable of `FlowFileTransformResult` objects, each of which indicates the Relationship the FlowFile should be transferred to, +the contents of the FlowFile, and any attributes that should be added to the FlowFile (or overwritten). The +`relationship` is a required argument. If the contents of the FlowFile are not to be written, +the `contents` should be unspecified or should be specified as `None`. Also it is more efficient to omit the `attributes` unless there is any attribute to add. + + [[process-context]] ==== context @@ -356,6 +401,56 @@ When there is nothing to return, it might be useful to yield the processor's res to run for the period of time defined by the processor's Yield Duration. This can be achieved by calling `context.yield_resources()` from the processor's `create` method right before returning `None`. +[[flowfile-source]] +=== MultipleFlowFileSource + +The `FlowFileSource` API provides a mechanism for creating FlowFiles and routing them based on their textual or binary contents. + +In order to implement the `MultipleFlowFileSource` API, a Python class must extend from the `nifiapi.MultipleFlowFileSource` class +and implement the `create(ProcessContext)` method, which returns an iterable of `FlowFileSourceResult` objects. Notice, that the difference between +`MultipleFlowFileSource's create(ProcessContext)` and other transform methods is +that the former does not expect an InputFlowFile object. That is because processors based on the `MultipleFlowFileSource` API +are "source" processors that do not accept incoming connections but are capable of creating FlowFiles themselves. + +Implementing a Processor based on `MultipleFlowFileSource` is very similar to implementing one based on `FlowFileTransformMultipleOutput`. +A simple implementation looks like this: + +---- +from nifiapi.flowfilesource import MultipleFlowFileSource, FlowFileSourceResult + +class CreateFlowFiles(MultipleFlowFileSource): + class Java: + implements = ['org.apache.nifi.python.processor.FlowFileSource'] + + class ProcessorDetails: + version = '0.0.1-SNAPSHOT' + description = '''A Python processor that creates FlowFiles.''' + + def __init__(self, **kwargs): + pass + + def create(self, context): + results = [] + results.append(FlowFileTransformResult(relationship = "success", contents = "Hello from Flowfile 1!", attributes = {"greeting": "hello"})) + results.append(FlowFileTransformResult(relationship = "success", contents = "Goodbye from Flowfile 2!", attributes = {"greeting": "goodbye"})) + + return results +---- + +As mentioned above, the `create` method only takes one argument: the context (of type `nifiapi.properties.ProcessContext`). + +The return type is an iterable of `FlowFileSourceResult` objects, each of which indicates the Relationship the FlowFile should be transferred to, +any attributes that should be added to the FlowFile and the contents of the FlowFile. The `relationship` is a required argument. +Each processor based on the `MultipleFlowFileSource` API has a `success` relationship and additional relationships can be +created in the Processor's Python code. `attributes` and `contents` are both optional. If `attributes` is not provided, +the FlowFile will still have the usual `filename`, `path` and `uuid` attributes, but no additional ones. +If `contents` is not provided, a FlowFile with no contents (only attributes) will be created. +In case there is no useful information to return from the `create` method, `return None` can be used instead of returning an +empty `FlowFileSourceResult`. When `create()` returns with `None`, the processor does not produce any output. +When there is nothing to return, it might be useful to yield the processor's resources and not schedule the processor +to run for the period of time defined by the processor's Yield Duration. This can be achieved by calling +`context.yield_resources()` from the processor's `create` method right before returning `None`. + [[property-descriptors]] === PropertyDescriptors diff --git a/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java index 834b5492d496..a339ef896e7f 100644 --- a/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java +++ b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java @@ -30,7 +30,11 @@ import org.apache.nifi.python.processor.FlowFileSource; import org.apache.nifi.python.processor.FlowFileSourceProxy; import org.apache.nifi.python.processor.FlowFileTransform; +import org.apache.nifi.python.processor.FlowFileTransformMultipleOutput; +import org.apache.nifi.python.processor.FlowFileTransformMultipleOutputProxy; import org.apache.nifi.python.processor.FlowFileTransformProxy; +import org.apache.nifi.python.processor.MultipleFlowFileSource; +import org.apache.nifi.python.processor.MultipleFlowFileSourceProxy; import org.apache.nifi.python.processor.PythonProcessorBridge; import org.apache.nifi.python.processor.RecordTransform; import org.apache.nifi.python.processor.RecordTransformProxy; @@ -162,9 +166,15 @@ public AsyncLoadedProcessor createProcessor(final String identifier, final Strin if (FlowFileTransform.class.getName().equals(implementedInterface)) { return new FlowFileTransformProxy(type, processorBridgeFactory, initialize); } + if (FlowFileTransformMultipleOutput.class.getName().equals(implementedInterface)) { + return new FlowFileTransformMultipleOutputProxy(type, processorBridgeFactory, initialize); + } if (RecordTransform.class.getName().equals(implementedInterface)) { return new RecordTransformProxy(type, processorBridgeFactory, initialize); } + if (MultipleFlowFileSource.class.getName().equals(implementedInterface)) { + return new MultipleFlowFileSourceProxy(type, processorBridgeFactory, initialize); + } if (FlowFileSource.class.getName().equals(implementedInterface)) { return new FlowFileSourceProxy(type, processorBridgeFactory, initialize); } diff --git a/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/FlowFileTransformMultipleOutput.java b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/FlowFileTransformMultipleOutput.java new file mode 100644 index 000000000000..9be35caabbb7 --- /dev/null +++ b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/FlowFileTransformMultipleOutput.java @@ -0,0 +1,26 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.python.processor; + +import java.util.List; + +public interface FlowFileTransformMultipleOutput extends PythonProcessor { + + List transformFlowFile(InputFlowFile flowFile); + +} diff --git a/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/FlowFileTransformMultipleOutputProxy.java b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/FlowFileTransformMultipleOutputProxy.java new file mode 100644 index 000000000000..59132956e01c --- /dev/null +++ b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/FlowFileTransformMultipleOutputProxy.java @@ -0,0 +1,93 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.python.processor; + +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import py4j.Py4JNetworkException; + +import java.util.List; +import java.util.Map; +import java.util.function.Supplier; + +@InputRequirement(Requirement.INPUT_REQUIRED) +public class FlowFileTransformMultipleOutputProxy extends PythonProcessorProxy { + + public FlowFileTransformMultipleOutputProxy(final String processorType, final Supplier bridgeFactory, final boolean initialize) { + super(processorType, bridgeFactory, initialize); + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + FlowFile flowFile = session.get(); + if (flowFile == null) { + return; + } + + final List results; + try (final StandardInputFlowFile inputFlowFile = new StandardInputFlowFile(session, flowFile)) { + results = getTransform().transformFlowFile(inputFlowFile); + } catch (final Py4JNetworkException e) { + throw new ProcessException("Failed to communicate with Python Process", e); + } catch (final Exception e) { + getLogger().error("Failed to transform {}", flowFile, e); + session.transfer(flowFile, REL_FAILURE); + return; + } + + for (final FlowFileTransformResult result : results) { + try { + FlowFile outputFlowFile = session.create(); + final String relationshipName = result.getRelationship(); + final Relationship relationship = new Relationship.Builder().name(relationshipName).build(); + final Map attributes = result.getAttributes(); + + if (REL_FAILURE.getName().equals(relationshipName)) { + if (attributes != null) { + outputFlowFile = session.putAllAttributes(flowFile, attributes); + } + + session.transfer(outputFlowFile, REL_FAILURE); + return; + } + + outputFlowFile = session.putAllAttributes( + (attributes != null) ? outputFlowFile : flowFile, + attributes); + + + final byte[] contents = result.getContents(); + if (contents != null) { + outputFlowFile = session.write(outputFlowFile, out -> out.write(contents)); + } + + session.transfer(outputFlowFile, relationship); + + } finally { + result.free(); + } + } + session.transfer(flowFile, REL_ORIGINAL); + } + +} diff --git a/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/MultipleFlowFileSource.java b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/MultipleFlowFileSource.java new file mode 100644 index 000000000000..d6ca738c08f1 --- /dev/null +++ b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/MultipleFlowFileSource.java @@ -0,0 +1,26 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.python.processor; + +import java.util.List; + +public interface MultipleFlowFileSource extends PythonProcessor { + + List createFlowFiles(); + +} diff --git a/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/MultipleFlowFileSourceProxy.java b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/MultipleFlowFileSourceProxy.java new file mode 100644 index 000000000000..b51d614fa44f --- /dev/null +++ b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/MultipleFlowFileSourceProxy.java @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.python.processor; + +import org.apache.nifi.annotation.behavior.DefaultRunDuration; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.SupportsBatching; +import org.apache.nifi.annotation.configuration.DefaultSchedule; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.scheduling.SchedulingStrategy; +import py4j.Py4JNetworkException; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Supplier; + +@InputRequirement(Requirement.INPUT_FORBIDDEN) +@SupportsBatching(defaultDuration = DefaultRunDuration.NO_BATCHING) +@DefaultSchedule(strategy = SchedulingStrategy.TIMER_DRIVEN, period = "1 min") +public class MultipleFlowFileSourceProxy extends PythonProcessorProxy { + + protected static final Relationship REL_SUCCESS = new Relationship.Builder() + .name("success") + .description("FlowFiles created by this processor can be routed to this relationship.") + .build(); + + private static final Set implicitRelationships = Set.of(REL_SUCCESS); + + public MultipleFlowFileSourceProxy(final String processorType, final Supplier bridgeFactory, final boolean initialize) { + super(processorType, bridgeFactory, initialize); + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + final List results; + try { + MultipleFlowFileSource transform = getTransform(); + results = transform.createFlowFiles(); + if (results == null) { + return; + } + } catch (final Py4JNetworkException e) { + throw new ProcessException("Failed to communicate with Python Process", e); + } catch (final Exception e) { + getLogger().error("Failed to create FlowFile", e); + return; + } + + for (final FlowFileSourceResult result : results) { + try { + final String relationshipName = result.getRelationship(); + final Relationship relationship = new Relationship.Builder().name(relationshipName).build(); + final Map attributes = result.getAttributes(); + final byte[] contents = result.getContents(); + + FlowFile output = createFlowFile(session, attributes, contents); + + if (REL_SUCCESS.getName().equals(relationshipName)) { + session.transfer(output, REL_SUCCESS); + } else { + session.transfer(output, relationship); + } + + } finally { + result.free(); + } + } + } + + protected FlowFile createFlowFile(final ProcessSession session, final Map attributes, final byte[] contents) { + FlowFile flowFile = session.create(); + if (attributes != null) { + flowFile = session.putAllAttributes(flowFile, attributes); + } + if (contents != null) { + flowFile = session.write(flowFile, out -> out.write(contents)); + } + return flowFile; + } + + @Override + protected Set getImplicitRelationships() { + return implicitRelationships; + } +} diff --git a/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-integration-tests/src/test/java/org.apache.nifi.py4j/PythonControllerInteractionIT.java b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-integration-tests/src/test/java/org.apache.nifi.py4j/PythonControllerInteractionIT.java index e696943f94c2..9d5f41fb70da 100644 --- a/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-integration-tests/src/test/java/org.apache.nifi.py4j/PythonControllerInteractionIT.java +++ b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-integration-tests/src/test/java/org.apache.nifi.py4j/PythonControllerInteractionIT.java @@ -756,6 +756,42 @@ public void testStateManagerExceptionHandling() { runner.getFlowFilesForRelationship("success").getFirst().assertAttributeEquals("exception_msg", "Set state failed"); } + @Test + public void testGeneratePagesCreatesMultipleFlowFiles() { + final TestRunner runner = createProcessor("GeneratePages"); + waitForValid(runner); + runner.run(); + + runner.assertTransferCount("success", 3); + + final List outputs = runner.getFlowFilesForRelationship("success"); + for (int i = 0; i < outputs.size(); i++) { + final MockFlowFile flowFile = outputs.get(i); + flowFile.assertContentEquals("page-" + i); + flowFile.assertAttributeEquals("page.index", String.valueOf(i)); + } + } + + @Test + public void testSplitLinesCreatesMultipleFlowFiles() { + final TestRunner runner = createFlowFileTransform("SplitLines"); + runner.enqueue("alpha\nbeta\ngamma"); + + runner.run(); + + runner.assertTransferCount("original", 1); + runner.assertTransferCount("success", 3); + + final List outputs = runner.getFlowFilesForRelationship("success"); + outputs.get(0).assertContentEquals("alpha"); + outputs.get(0).assertAttributeEquals("split.index", "0"); + outputs.get(1).assertContentEquals("beta"); + outputs.get(1).assertAttributeEquals("split.index", "1"); + outputs.get(2).assertContentEquals("gamma"); + outputs.get(2).assertAttributeEquals("split.index", "2"); + } + + public interface StringLookupService extends ControllerService { Optional lookup(Map coordinates); } diff --git a/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-python-test-extensions/src/main/resources/extensions/GeneratePages.py b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-python-test-extensions/src/main/resources/extensions/GeneratePages.py new file mode 100644 index 000000000000..36f6aaa5a10c --- /dev/null +++ b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-python-test-extensions/src/main/resources/extensions/GeneratePages.py @@ -0,0 +1,37 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from nifiapi.multipleflowfilesource import MultipleFlowFileSource, FlowFileSourceResult + + +class GeneratePages(MultipleFlowFileSource): + class Java: + implements = ['org.apache.nifi.python.processor.MultipleFlowFileSource'] + + class ProcessorDetails: + version = '0.0.1-SNAPSHOT' + description = 'Emits a FlowFile for each page encountered when paging through results.' + tags = ['pagination', 'test', 'python'] + + def __init__(self, **kwargs): + super().__init__() + + def create(self, context): + results = [] + for index in range(3): + attributes = {'page.index': str(index)} + contents = f'page-{index}' + results.append(FlowFileSourceResult(relationship='success', attributes=attributes, contents=contents)) + return results \ No newline at end of file diff --git a/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-python-test-extensions/src/main/resources/extensions/SplitLines.py b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-python-test-extensions/src/main/resources/extensions/SplitLines.py new file mode 100644 index 000000000000..7fcce3e82fb6 --- /dev/null +++ b/nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-python-test-extensions/src/main/resources/extensions/SplitLines.py @@ -0,0 +1,43 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from nifiapi.flowfiletransformmultipleoutput import FlowFileTransformMultipleOutput, FlowFileTransformResult + + +class SplitLines(FlowFileTransformMultipleOutput): + class Java: + implements = ['org.apache.nifi.python.processor.FlowFileTransformMultipleOutput'] + + class ProcessorDetails: + version = '0.0.1-SNAPSHOT' + description = 'Splits the incoming FlowFile into one FlowFile per line of text.' + tags = ['split', 'line', 'test', 'python'] + + def __init__(self, **kwargs): + super().__init__() + + def transform(self, context, flowFile): + contents = flowFile.getContentsAsBytes().decode('utf-8') + if not contents: + return [] + + results = [] + for index, line in enumerate(contents.splitlines()): + attributes = { + 'split.index': str(index), + 'split.line.length': str(len(line)) + } + results.append(FlowFileTransformResult(relationship='success', attributes=attributes, contents=line)) + return results \ No newline at end of file diff --git a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfilesource.py b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfilesource.py index d5d88de21e6b..2c2e7502e06a 100644 --- a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfilesource.py +++ b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfilesource.py @@ -16,7 +16,7 @@ from abc import ABC, abstractmethod from nifiapi.__jvm__ import JvmHolder from nifiapi.properties import ProcessContext - +from nifiapi.flowfilesourceresult import FlowFileSourceResult class FlowFileSource(ABC): # These will be set by the PythonProcessorAdapter when the component is created @@ -36,30 +36,3 @@ def createFlowFile(self): def create(self, context): pass -class FlowFileSourceResult: - class Java: - implements = ['org.apache.nifi.python.processor.FlowFileSourceResult'] - - def __init__(self, relationship, attributes = None, contents = None): - self.relationship = relationship - self.attributes = attributes - if contents is not None and isinstance(contents, str): - self.contents = str.encode(contents) - else: - self.contents = contents - - def getRelationship(self): - return self.relationship - - def getContents(self): - return self.contents - - def getAttributes(self): - if self.attributes is None: - return None - - map = JvmHolder.jvm.java.util.HashMap() - for key, value in self.attributes.items(): - map.put(key, value) - - return map diff --git a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfilesourceresult.py b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfilesourceresult.py new file mode 100644 index 000000000000..ba0a416abc6f --- /dev/null +++ b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfilesourceresult.py @@ -0,0 +1,44 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from nifiapi.__jvm__ import JvmHolder + +class FlowFileSourceResult: + class Java: + implements = ['org.apache.nifi.python.processor.FlowFileSourceResult'] + + def __init__(self, relationship, attributes = None, contents = None): + self.relationship = relationship + self.attributes = attributes + if contents is not None and isinstance(contents, str): + self.contents = str.encode(contents) + else: + self.contents = contents + + def getRelationship(self): + return self.relationship + + def getContents(self): + return self.contents + + def getAttributes(self): + if self.attributes is None: + return None + + map = JvmHolder.jvm.java.util.HashMap() + for key, value in self.attributes.items(): + map.put(key, value) + + return map diff --git a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfiletransform.py b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfiletransform.py index 4b2f47d17e6f..2da7d976e8ce 100644 --- a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfiletransform.py +++ b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfiletransform.py @@ -16,7 +16,7 @@ from abc import ABC, abstractmethod from nifiapi.__jvm__ import JvmHolder from nifiapi.properties import ProcessContext - +from nifiapi.flowfiletransformresult import FlowFileTransformResult class FlowFileTransform(ABC): # These will be set by the PythonProcessorAdapter when the component is created @@ -37,30 +37,3 @@ def transform(self, context, flowFile): pass -class FlowFileTransformResult: - class Java: - implements = ['org.apache.nifi.python.processor.FlowFileTransformResult'] - - def __init__(self, relationship, attributes = None, contents = None): - self.relationship = relationship - self.attributes = attributes - if contents is not None and isinstance(contents, str): - self.contents = str.encode(contents) - else: - self.contents = contents - - def getRelationship(self): - return self.relationship - - def getContents(self): - return self.contents - - def getAttributes(self): - if self.attributes is None: - return None - - map = JvmHolder.jvm.java.util.HashMap() - for key, value in self.attributes.items(): - map.put(key, value) - - return map \ No newline at end of file diff --git a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfiletransformmultipleoutput.py b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfiletransformmultipleoutput.py new file mode 100644 index 000000000000..192a96e6ca44 --- /dev/null +++ b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfiletransformmultipleoutput.py @@ -0,0 +1,41 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from abc import ABC, abstractmethod +from nifiapi.__jvm__ import JvmHolder +from nifiapi.properties import ProcessContext +from nifiapi.flowfiletransformresult import FlowFileTransformResult + + +class FlowFileTransformMultipleOutput(ABC): + # These will be set by the PythonProcessorAdapter when the component is created + identifier = None + logger = None + + def __init__(self): + self.arrayList = JvmHolder.jvm.java.util.ArrayList + + def setContext(self, context): + self.process_context = ProcessContext(context) + + def transformFlowFile(self, flowfile): + results = self.arrayList() + for (result) in self.transform(self.process_context, flowfile): + results.add(result) + return results + + @abstractmethod + def transform(self, context, flowFile): + pass diff --git a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfiletransformresult.py b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfiletransformresult.py new file mode 100644 index 000000000000..b176567fd1aa --- /dev/null +++ b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/flowfiletransformresult.py @@ -0,0 +1,44 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from nifiapi.__jvm__ import JvmHolder + +class FlowFileTransformResult: + class Java: + implements = ['org.apache.nifi.python.processor.FlowFileTransformResult'] + + def __init__(self, relationship, attributes = None, contents = None): + self.relationship = relationship + self.attributes = attributes + if contents is not None and isinstance(contents, str): + self.contents = str.encode(contents) + else: + self.contents = contents + + def getRelationship(self): + return self.relationship + + def getContents(self): + return self.contents + + def getAttributes(self): + if self.attributes is None: + return None + + map = JvmHolder.jvm.java.util.HashMap() + for key, value in self.attributes.items(): + map.put(key, value) + + return map \ No newline at end of file diff --git a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/multipleflowfilesource.py b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/multipleflowfilesource.py new file mode 100644 index 000000000000..2a7310570fa8 --- /dev/null +++ b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-extension-api/src/main/python/src/nifiapi/multipleflowfilesource.py @@ -0,0 +1,41 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from abc import ABC, abstractmethod +from nifiapi.__jvm__ import JvmHolder +from nifiapi.properties import ProcessContext +from nifiapi.flowfilesourceresult import FlowFileSourceResult + + +class MultipleFlowFileSource(ABC): + # These will be set by the PythonProcessorAdapter when the component is created + identifier = None + logger = None + + def __init__(self): + self.arrayList = JvmHolder.jvm.java.util.ArrayList + + def setContext(self, context): + self.process_context = ProcessContext(context) + + def createFlowFiles(self): + results = self.arrayList() + for (result) in self.create(self.process_context): + results.add(result) + return results + + @abstractmethod + def create(self, context): + pass diff --git a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-framework/src/main/python/framework/ExtensionManager.py b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-framework/src/main/python/framework/ExtensionManager.py index a6bb03f4ac94..a4658dcb22d4 100644 --- a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-framework/src/main/python/framework/ExtensionManager.py +++ b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-framework/src/main/python/framework/ExtensionManager.py @@ -55,7 +55,9 @@ class ExtensionManager: """ processor_interfaces = ['org.apache.nifi.python.processor.FlowFileTransform', + 'org.apache.nifi.python.processor.FlowFileTransformMultipleOutput', 'org.apache.nifi.python.processor.RecordTransform', + 'org.apache.nifi.python.processor.MultipleFlowFileSource', 'org.apache.nifi.python.processor.FlowFileSource'] processor_details = {} processor_class_by_name = {} diff --git a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-framework/src/main/python/framework/ProcessorInspection.py b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-framework/src/main/python/framework/ProcessorInspection.py index 0eddaad0a7b1..cfa5e5375191 100644 --- a/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-framework/src/main/python/framework/ProcessorInspection.py +++ b/nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-framework/src/main/python/framework/ProcessorInspection.py @@ -23,7 +23,9 @@ import ExtensionDetails PROCESSOR_INTERFACES = ['org.apache.nifi.python.processor.FlowFileTransform', + 'org.apache.nifi.python.processor.FlowFileTransformMultipleOutput', 'org.apache.nifi.python.processor.RecordTransform', + 'org.apache.nifi.python.processor.MultipleFlowFileSource', 'org.apache.nifi.python.processor.FlowFileSource'] logger = logging.getLogger("python.ProcessorInspection")