【NIFI】 开发自定义Nifi Processor

  1 /*
  2  * Licensed to the Apache Software Foundation (ASF) under one or more
  3  * contributor license agreements.  See the NOTICE file distributed with
  4  * this work for additional information regarding copyright ownership.
  5  * The ASF licenses this file to You under the Apache License, Version 2.0
  6  * (the "License"); you may not use this file except in compliance with
  7  * the License.  You may obtain a copy of the License at
  8  *
  9  *     http://www.apache.org/licenses/LICENSE-2.0
 10  *
 11  * Unless required by applicable law or agreed to in writing, software
 12  * distributed under the License is distributed on an "AS IS" BASIS,
 13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 14  * See the License for the specific language governing permissions and
 15  * limitations under the License.
 16  */
 17 package com.test.processors;
 18 
 19 import org.apache.nifi.components.PropertyDescriptor;
 20 import org.apache.nifi.flowfile.FlowFile;
 21 import org.apache.nifi.annotation.behavior.ReadsAttribute;
 22 import org.apache.nifi.annotation.behavior.ReadsAttributes;
 23 import org.apache.nifi.annotation.behavior.WritesAttribute;
 24 import org.apache.nifi.annotation.behavior.WritesAttributes;
 25 import org.apache.nifi.annotation.lifecycle.OnScheduled;
 26 import org.apache.nifi.annotation.documentation.CapabilityDescription;
 27 import org.apache.nifi.annotation.documentation.SeeAlso;
 28 import org.apache.nifi.annotation.documentation.Tags;
 29 import org.apache.nifi.processor.exception.ProcessException;
 30 import org.apache.nifi.processor.AbstractProcessor;
 31 import org.apache.nifi.processor.ProcessContext;
 32 import org.apache.nifi.processor.ProcessSession;
 33 import org.apache.nifi.processor.ProcessorInitializationContext;
 34 import org.apache.nifi.processor.Relationship;
 35 import org.apache.nifi.processor.util.StandardValidators;
 36 
 37 import java.io.InputStreamReader;
 38 import java.io.StringWriter;
 39 import java.util.ArrayList;
 40 import java.util.Collections;
 41 import java.util.HashSet;
 42 import java.util.List;
 43 import java.util.Set;
 44 import java.util.concurrent.atomic.AtomicReference;
 45 
 46 @Tags({"example"})
 47 @CapabilityDescription("Provide a description")
 48 @SeeAlso({})
 49 @ReadsAttributes({@ReadsAttribute(attribute="", description="")})
 50 @WritesAttributes({@WritesAttribute(attribute="", description="")})
 51 public class MyProcessor extends AbstractProcessor {
 52 
 53     public static final PropertyDescriptor MY_PROPERTY = new PropertyDescriptor
 54             .Builder().name("MY_PROPERTY")
 55             .displayName("My property")
 56             .description("Example Property")
 57             .required(true)
 58             .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
 59             .build();
 60 
 61     public static final Relationship MY_RELATIONSHIP_SUCCESS = new Relationship.Builder()
 62             .name("sucess")
 63             .description("Example relationship Success")
 64             .build();
 65     
 66     public static final Relationship MY_RELATIONSHIP_FAILURE = new Relationship.Builder()
 67             .name("failure")
 68             .description("Example relationship Failure")
 69             .build();
 70 
 71     private List<PropertyDescriptor> descriptors;
 72 
 73     private Set<Relationship> relationships;
 74 
 75     @Override
 76     protected void init(final ProcessorInitializationContext context) {
 77         final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
 78         descriptors.add(MY_PROPERTY);
 79         this.descriptors = Collections.unmodifiableList(descriptors);
 80 
 81         final Set<Relationship> relationships = new HashSet<Relationship>();
 82         relationships.add(MY_RELATIONSHIP_SUCCESS);
 83         relationships.add(MY_RELATIONSHIP_FAILURE);
 84         this.relationships = Collections.unmodifiableSet(relationships);
 85     }
 86 
 87     @Override
 88     public Set<Relationship> getRelationships() {
 89         return this.relationships;
 90     }
 91 
 92     @Override
 93     public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
 94         return descriptors;
 95     }
 96 
 97     @OnScheduled
 98     public void onScheduled(final ProcessContext context) {
 99 
100     }
101 
102     @Override
103     public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
104         FlowFile flowFile = session.get();
105         if ( flowFile == null ) {
106             return;
107         }
108         // TODO implement
109         final AtomicReference<String> value = new AtomicReference<>();
110         session.read(flowFile, in -> {
111             try{
112                 StringWriter sw = new StringWriter();
113                 InputStreamReader inr = new InputStreamReader(in);
114                 char[] buffer = new char[1024];
115                 int n = 0;
116                 while (-1 != (n = inr.read(buffer))) {
117                     sw.write(buffer, 0, n);
118                 }
119                 String str = sw.toString();
120                 
121                 String result = "处理了：" + str + context.getProperty("MY_PROPERTY").getValue();
122                 value.set(result);
123             }catch(Exception ex){
124                 ex.printStackTrace();
125                 getLogger().error("Failed to read json string.");
126             }
127         });
128 
129         String results = value.get();
130         if(results != null && !results.isEmpty()){
131             flowFile = session.putAttribute(flowFile, "match", results);
132         }
133 
134         flowFile = session.write(flowFile, out -> out.write(value.get().getBytes()));
135 
136         session.transfer(flowFile, MY_RELATIONSHIP_SUCCESS);
137         
138     }
139 }