以下是一个使用Apache NIFI执行流命令的示例代码：

首先，需要在NIFI中创建一个ExecuteStreamCommand处理器，然后配置它以执行所需的命令。

# ExecuteStreamCommand配置参数
Command Path: /path/to/command.sh
Command Arguments: --arg1 value1 --arg2 value2

然后，您可以使用Java编写一个简单的NIFI Processor来执行命令并处理输出。以下是一个示例代码：

import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
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.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processor.AbstractProcessor;

import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

@Tags({"example"})
@SupportsBatching
@InputRequirement(Requirement.INPUT_REQUIRED)
@CapabilityDescription("Example NIFI processor to execute a stream command")
@WritesAttributes({
        @WritesAttribute(attribute = "example.output", description = "Example output attribute")
})
public class ExecuteStreamCommandProcessor extends AbstractProcessor {

    public static final PropertyDescriptor COMMAND_PATH = new PropertyDescriptor
            .Builder().name("Command Path")
            .description("Command path")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor COMMAND_ARGUMENTS = new PropertyDescriptor
            .Builder().name("Command Arguments")
            .description("Command arguments")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final Relationship SUCCESS = new Relationship.Builder()
            .name("success")
            .description("Success relationship")
            .build();

    public static final Relationship FAILURE = new Relationship.Builder()
            .name("failure")
            .description("Failure relationship")
            .build();

    private List descriptors;

    private Set relationships;

    @Override
    protected void init(final ProcessorInitializationContext context) {
        final List descriptors = new ArrayList();
        descriptors.add(COMMAND_PATH);
        descriptors.add(COMMAND_ARGUMENTS);
        this.descriptors = Collections.unmodifiableList(descriptors);

        final Set relationships = new HashSet();
        relationships.add(SUCCESS);
        relationships.add(FAILURE);
        this.relationships = Collections.unmodifiableSet(relationships);
    }

    @Override
    public Set getRelationships() {
        return this.relationships;
    }

    @Override
    public final List getSupportedPropertyDescriptors() {
        return descriptors;
    }

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
        FlowFile flowFile = session.get();
        if (flowFile == null) {
            return;
        }

        // 获取配置参数
        String commandPath = context.getProperty(COMMAND_PATH).getValue();
        String commandArguments = context.getProperty(COMMAND_ARGUMENTS).getValue();

        try {
            // 执行命令
            ProcessBuilder processBuilder = new ProcessBuilder(commandPath, commandArguments);
            Process process = processBuilder.start();

            // 处理命令输出
            String output = readStream(process.getInputStream());

            // 处理命令错误输出
            String errorOutput = readStream(process.getErrorStream());

            // 如果命令执行成功，则将输出写入FlowFile属性
            if (process.waitFor() == 0) {
                flowFile = session.putAttribute(flowFile, "example.output", output);

                // 将FlowFile发送到成功关系
                session.transfer(flowFile, SUCCESS);
            } else {
                // 将FlowFile发送到失败关系
                session.transfer(flowFile, FAILURE);
            }
        } catch (Exception e) {
            getLogger().error("Failed to execute stream command", e);
            session.transfer(flowFile, FAILURE);
        }
    }

    private String readStream