使用myeclipse及java构建自己的nifi 自定义processor

悠悠 2022-04-11 13:00 432阅读 0赞

自己最近开始摸索使用eclipse及java构建自己自定义的nifi processor，并部署到nifi lib中使用，最终构建好的nifi template如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTE2MTY1NjAw_size_16_color_FFFFFF_t_70][]

具体如何构建nifi template在这里就不做重复写了，可以自己先摸索个简单的template 跑起来后在构建自己自定义的processor

好了，这里开始进入正题，开始上代码：

工程的结构图如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTE2MTY1NjAw_size_16_color_FFFFFF_t_70 1][]

1.定义自己的主要处理代码，我们在这里定义为changeDate，代码如下：

package com.nifi.test;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.io.OutputStream;
    import java.util.ArrayList;
    import java.util.Collections;
    import java.util.HashSet;
    import java.util.List;
    import java.util.Set;
    import java.util.concurrent.atomic.AtomicBoolean;
    import java.util.concurrent.atomic.AtomicReference;
    import org.apache.commons.io.IOUtils;
    import org.apache.commons.lang3.builder.ToStringBuilder;
    import org.apache.commons.lang3.builder.ToStringStyle;
    import org.apache.nifi.annotation.behavior.SideEffectFree;
    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.AbstractProcessor;
    import org.apache.nifi.processor.ProcessContext;
    import org.apache.nifi.processor.ProcessSession;
    import org.apache.nifi.processor.ProcessorInitializationContext;
    import org.apache.nifi.processor.Relationship;
    import org.apache.nifi.processor.exception.ProcessException;
    import org.apache.nifi.processor.io.InputStreamCallback;
    import org.apache.nifi.processor.io.OutputStreamCallback;
    import org.apache.nifi.processor.util.StandardValidators;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import com.fasterxml.jackson.core.JsonProcessingException;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.jayway.jsonpath.JsonPath;
    
    @SideEffectFree
    @Tags({"CheckDate","My Processor CheckDate"})
    @CapabilityDescription("to format date,from 'year' to 'month'")
    public class ChangeDate extends AbstractProcessor{
    	private List<PropertyDescriptor> properties;
    	private Set<Relationship> relationships;
    	private static final Logger log = LoggerFactory.getLogger(ChangeDate.class);
    
    	public static final PropertyDescriptor JSON_PATH = new PropertyDescriptor.Builder()
    	        .name("Json Path")
    	        .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();
    	
    	
    	@Override
    	public void init(final ProcessorInitializationContext context){
    	    List<PropertyDescriptor> properties = new ArrayList<PropertyDescriptor>();
    	    properties.add(JSON_PATH);
    	    this.properties = Collections.unmodifiableList(properties);
    
    	    Set<Relationship> relationships = new HashSet<Relationship>();
    	    relationships.add(SUCCESS);
    	  
    	    relationships.add(FAILURE);
    	    this.relationships = Collections.unmodifiableSet(relationships);
    	}
    
    	@Override
    	public Set<Relationship> getRelationships(){
    	    return relationships;
    	}
    
    	@Override
    	public List<PropertyDescriptor> getSupportedPropertyDescriptors(){
    	    return properties;
    	}
    	@Override
    	public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
    
    	    FlowFile flowfile = session.get();
    	    final AtomicReference<String> jsonIn = new AtomicReference<String>() ;
    	    final AtomicReference<String> jsonOutAtom = new AtomicReference<String>() ;
    	    
    	    
    	    final AtomicReference<FlowMsgOut> msgOut = new AtomicReference<FlowMsgOut>() ;
    	    final AtomicReference<String> checkDate = new AtomicReference<String>();
    	    final AtomicBoolean failure = new AtomicBoolean(false);
    	    
    	    session.read(flowfile, new InputStreamCallback() {
    	    	
    	    	@Override
    	        public void process(InputStream in) throws IOException {
    	        	
    	            //try{
    	            	
    	            	//将输入流转换成对象
    	            	/*
    	                String json = "{\"phoneId\": \"111\"," + 
    	                		"    \"phoneName\": \"huawei\"," + 
    	                		"    \"phonePrice\": \"6856\"," + 
    	                		"    \"repaymentPeriod\": \"2年\"}";
    	                		*/
    	                		
    	            	getLogger().info("begin get inputStream and exchange this for json");
    	            	//getLogger().info("inputStream is:",ToStringBuilder.reflectionToString(in, ToStringStyle.SHORT_PREFIX_STYLE));
    	            	getLogger().info("current input is null:"+String.valueOf(in == null));
    	            	
    	                String json = IOUtils.toString(in); 
    	            	
    	            	
    	            	
    	                getLogger().info("current in String value is:"+json);
    	                //jsonIn.set(json);
    	                //log.info("RECEIVE :"+json);
    	                System.out.println("RECEIVE:" + json); 
    	                
    	                
    	                FlowMsgIn msgIn = getFlowMsg(json) ;
    	                
    	                //业务逻辑
    	                checkDate.set(new Rules().newDate(msgIn)) ;
    	                //checkDate.set("10") ;
    	                
    	                //构造输出对象
    	                FlowMsgOut out = new FlowMsgOut() ;
    	                out.setPhoneId(msgIn.getPhoneId());
    	                out.setPhoneName(msgIn.getPhoneName());
    	                out.setPhonePrice(msgIn.getPhonePrice());
    	                out.setRepaymentPeriodProcessed(checkDate.get());
    //	                out.setPhoneId("111");
    //	                out.setPhoneName("华为");
    //	                out.setPhonePrice("5555");
    //	                out.setRepaymentPeriodProcessed("10");
    	                msgOut.set(out) ;
    	            /*
    	            }catch(Exception ex){
    	                ex.printStackTrace();
    	                getLogger().error("Failed in InputStreamCallback.");
    	                failure.set(true) ;
    	            }
    	            */
    	        }
    	    });
    	    
    	    ObjectMapper objectMapper = new ObjectMapper();  
    	    String jsonOut = "" ;
    		try {
    			jsonOut = objectMapper.writeValueAsString(msgOut.get());
    		} catch (JsonProcessingException e) {
    			e.printStackTrace();
    			failure.set(true) ;
    		} 
    		jsonOutAtom.set(jsonOut);
    		
    		//设置输出流内容
    		if(failure != null && failure.get()){
    		
    			flowfile = session.putAttribute(flowfile,"jsonIn",jsonIn.get());
    		}else{
    			flowfile = session.putAttribute(flowfile,"jsonOut",jsonOut);
    		}
    		//flowfile = session.putAttribute(flowfile,"jsonOut",jsonOut);
    		
    	    //写输出流
    	    flowfile = session.write(flowfile, new OutputStreamCallback() {
    
    			@Override
    			public void process(OutputStream out) throws IOException {
    				// TODO Auto-generated method stub
    				out.write(jsonOutAtom.get().getBytes());
    				
    			}
    
    //	        public void process(OutputStream out) throws IOException {
    //	        	out.write(jsonIn.get().getBytes());
    //	        	
    //	        }
    	    });
    	    //设置流向
    	    if(failure.get()){
    	    	
    	    	session.transfer(flowfile, FAILURE);
    	    }else{
    	    	
    	    	session.transfer(flowfile, SUCCESS) ;
    	    }
    	    
    	    //session.transfer(flowfile, SUCCESS) ;
    	    
    	   // 
    	}
    	
    	public FlowMsgIn getFlowMsg(String flowJson){
    		
    		FlowMsgIn msg = new FlowMsgIn() ;
    		
    		//getLogger().info("current flowJson is:"+flowJson);
    		msg.setRepaymentPeriod((String) JsonPath.read(flowJson, "$.repaymentPeriod"));
    		msg.setPhoneId((String) JsonPath.read(flowJson, "$.phoneId"));
    		msg.setPhoneName((String) JsonPath.read(flowJson, "$.phoneName"));
    		msg.setPhonePrice((String) JsonPath.read(flowJson, "$.phonePrice"));
    
    		
    		return msg ; 
    	}
    	
    	
    	public static void main(String[] args) {
    		ChangeDate d = new ChangeDate();
    		String flowJson = "{\"phoneId\":\"111\",\"phoneName\":\"华为\",\"phonePrice\":\"6856\",\"repaymentPeriod\":\"1年\"}";
    		FlowMsgIn msgIn = d.getFlowMsg(flowJson);
    		System.out.println(msgIn.toString());
    		//d.getFlowMsg(flowJson);
    		//final AtomicReference<String> checkDate = new AtomicReference<String>();
    		//checkDate.set(new Rules().newDate(msgIn)) ;
    		//System.out.println(ToStringBuilder.reflectionToString(checkDate.get().getBytes(), ToStringStyle.SHORT_PREFIX_STYLE));
    	}
    	
    	@Override
        public String toString () {
            return ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE);
        } 
    	
    
    }
    
    
    
    
    
    /*
     输入流示例
     {"phoneId":"111","phoneName":"华为","phonePrice":"6856","repaymentPeriod":"1年"}
     {"phoneId":"111","phoneName":"华为","phonePrice":"6856","repaymentPeriod":"24月"}
    */
    
    /*
    输入流示例
     {"phoneId":"111","phoneName":"华为","phonePrice":"6856","repaymentPeriodProcessed":"12月"}
     {"phoneId":"111","phoneName":"华为","phonePrice":"6856","repaymentPeriod":"24月"}
    
    */

这里面的代码，主要是要使用onTrigger()类，这里面才是自己处理数据的具体方法，放在这里面，

其中

@SideEffectFree  
@Tags(\{"CheckDate","My Processor CheckDate"\})  //这里的注解主要是使用在UI界面中 供检索到该processor使用  
@CapabilityDescription("to format date,from 'year' to 'month'") ///这里为该processor的description

2.其他DAO层代码及Rule代码如下：

package com.nifi.test;
    
    public class FlowMsgIn {
    
    	private String phoneId;
    	private String phoneName;
    	private String phonePrice;
    	private String repaymentPeriod;
    
    	public String getPhoneId() {
    		return phoneId;
    	}
    
    	public void setPhoneId(String phoneId) {
    		this.phoneId = phoneId;
    	}
    
    	public String getPhoneName() {
    		return phoneName;
    	}
    
    	public void setPhoneName(String phoneName) {
    		this.phoneName = phoneName;
    	}
    
    	public String getPhonePrice() {
    		return phonePrice;
    	}
    
    	public void setPhonePrice(String phonePrice) {
    		this.phonePrice = phonePrice;
    	}
    
    	public String getRepaymentPeriod() {
    		return repaymentPeriod;
    	}
    
    	public void setRepaymentPeriod(String repaymentPeriod) {
    		this.repaymentPeriod = repaymentPeriod;
    	}
    
    	@Override
    	public String toString() {
    		return "FlowMsgIn [phoneId=" + phoneId + ", phoneName=" + phoneName
    				+ ", phonePrice=" + phonePrice + ", repaymentPeriod="
    				+ repaymentPeriod + "]";
    	}
    }
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    *************************************************
    package com.nifi.test;
    
    public class FlowMsgOut {
    
    	private String phoneId;
    	private String phoneName;
    	private String phonePrice;
    	private String repaymentPeriod;
    	//private String repaymentPeriodProcessed ;
    
    	public String getPhoneId() {
    		return phoneId;
    	}
    
    	public void setPhoneId(String phoneId) {
    		this.phoneId = phoneId;
    	}
    
    	public String getPhoneName() {
    		return phoneName;
    	}
    
    	public void setPhoneName(String phoneName) {
    		this.phoneName = phoneName;
    	}
    
    	public String getPhonePrice() {
    		return phonePrice;
    	}
    
    	public void setPhonePrice(String phonePrice) {
    		this.phonePrice = phonePrice;
    	}
    
    	public String getRepaymentPeriodProcessed() {
    		return repaymentPeriod;
    	}
    
    	public void setRepaymentPeriodProcessed(String repaymentPeriod) {
    		this.repaymentPeriod = repaymentPeriod;
    	}
    
    	@Override
    	public String toString() {
    		return "FlowMsgOut [phoneId=" + phoneId + ", phoneName=" + phoneName
    				+ ", phonePrice=" + phonePrice + ", repaymentPeriod="
    				+ repaymentPeriod + "]";
    	}
    }
    
    
    
    ******************************************
    package com.nifi.test;
    
    /**
     * 调用已有的手机贷款信息数据
     * 判断贷款年限是否为月
     * 如为年则转成月
     */
    public class Rules {
    	public String newDate(FlowMsgIn msg){
    		
    		String str = msg.getRepaymentPeriod();
    		String a = str.substring(str.length() - 1);
    
    		if ("Y".equals(a)) {
    			String month = str.substring(0, str.length() - 1);
    			Integer temp = Integer.parseInt(month);
    			temp = temp * 12;
    			StringBuffer buffer = new StringBuffer();
    			buffer.append(temp.toString());
    			buffer.append("月");
    			return buffer.substring(0);
    		} else {
    			return str;
    		}
    	}
    }

这部分的代码只是简单地DAO层及简单的数据处理方式

其中还需要在src/main/resources/META-INF/services文件夹中新建一个 org.apache.nifi.processor.Processor 文件 里面的内容为需要暴露出来作为processor接口函数的类的名称，本工程的实例如下：

![20181206155619241.png][]

3.最终的结果：

通过以上代码生成的nar包 并放在nifi的lib文件中运行，最终的结果为，把json数据读取到了自定义的processor中，对年转换为月 在生成json 并最终输出到指定的文件中。

好啦 自己运行摸索吧，

附件POM文件配置：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    	<modelVersion>4.0.0</modelVersion>
    
    	<groupId>com.delta</groupId>
    	<artifactId>ChangeDate</artifactId>
    	<version>0.0.1</version>
    	<packaging>nar</packaging>
    
    	<name>nifi.json</name>
    	<url>http://maven.apache.org</url>
    
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    		<java.version>1.8</java.version>
    		<nifi.version>1.8.0</nifi.version>
    	</properties>
    
    	<dependencies>
    		
    			
    			
    			
    			
    		
    
    		
    			
    			
    			
                
    		
    
    		<dependency>
    			<groupId>org.apache.nifi</groupId>
    			<artifactId>nifi-utils</artifactId>
    			<version>${nifi.version}</version>
    			
    		</dependency>
    
    		
    			
    			
    			
    		
    
    		<dependency>
    			<groupId>org.apache.commons</groupId>
    			<artifactId>commons-io</artifactId>
    			<version>1.3.2</version>
    		</dependency>
    
    		<dependency>
    			<groupId>com.jayway.jsonpath</groupId>
    			<artifactId>json-path</artifactId>
    			<version>1.2.0</version>
    		</dependency>
    
    		
    			
    			
    			
    			
    		
    
    		
    			
    			
    			
    			
    		
    
    		<dependency>
    			<groupId>com.fasterxml.jackson.core</groupId>
    			<artifactId>jackson-databind</artifactId>
    			<version>2.8.3</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.apache.nifi</groupId>
    			<artifactId>nifi-properties</artifactId>
    			<version>${nifi.version}</version>
    			<scope>provided</scope>
    		</dependency>
    
    		<dependency>
        		<groupId>org.apache.nifi</groupId>
        		<artifactId>nifi-distributed-cache-client-service-api</artifactId>
        		<version>${nifi.version}</version>
    		</dependency>
    
    	</dependencies>
    
    	<build>
    		<plugins>
    			<plugin>
    				<groupId>org.apache.nifi</groupId>
    				<artifactId>nifi-nar-maven-plugin</artifactId>
    				<version>1.2.0</version>
    				<extensions>true</extensions>
    			</plugin>
    			 
    			<plugin>
    				<groupId>org.apache.maven.plugins</groupId>
    				<artifactId>maven-surefire-plugin</artifactId>
    				<version>2.19.1</version>
    				<configuration>
          				<skipTests>true</skipTests>
       				</configuration>
    			</plugin>
    
    		</plugins>
    	</build>
    </project>

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTE2MTY1NjAw_size_16_color_FFFFFF_t_70]: /images/20220411/1ed97088a43f4776a563917f986647bd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTE2MTY1NjAw_size_16_color_FFFFFF_t_70 1]: /images/20220411/a5376c8158734eb8978d13f7c8d84ca4.png
[20181206155619241.png]: /images/20220411/f6b237429e4d4805bc05135f9aca1010.png