• 欢迎访问开心洋葱网站,在线教程,推荐使用最新版火狐浏览器和Chrome浏览器访问本网站,欢迎加入开心洋葱 QQ群
  • 为方便开心洋葱网用户,开心洋葱官网已经开启复制功能!
  • 欢迎访问开心洋葱网站,手机也能访问哦~欢迎加入开心洋葱多维思维学习平台 QQ群
  • 如果您觉得本站非常有看点,那么赶紧使用Ctrl+D 收藏开心洋葱吧~~~~~~~~~~~~~!
  • 由于近期流量激增,小站的ECS没能经的起亲们的访问,本站依然没有盈利,如果各位看如果觉着文字不错,还请看官给小站打个赏~~~~~~~~~~~~~!

NiFi 学习 —自己实现处理器

NIFI supingemail 2189次浏览 0个评论

.

简单描述一下自己通过学习nifi和查看源码,然后自己通过nifi的体现很基础的实现自己的业务的处理器开发。

适合刚入门的看,有nifi基础的,可以跳过!

目录

一、 获取一个json文件的文本信息.

1.Json文档的输入端,需要一个GenerateFlowFile处理器,做如下配置:

2.拖拽自己实现的处理器

3.使用PutFile

二、合并文本的内容

1.Json文档的输入端

2.拖拽自己实现的处理器

3.使用PutFile

4.整体的流程图如下所示:

三、给一个文件内容添加头信息

1.Json文档的输入端

2.拖拽自己实现的处理器

3.使用PutFile

4.整体的流程图如下所示

一、 获取一个json文件的文本信息.

       假定我们需要在一段json文件中,获取json串中的某个key所对应的value的值,那么在nifi的架子上,该如何操作呐?共分为三步:

 

1.Json文档的输入端,需要一个GenerateFlowFile处理器,做如下配置:

Custom text 存放的是目标json串.json如下所示:{“error_code”:0,”reason”:”ok”,”result”:{“items”:[{“regStatus”:”存续”,”estiblishTime”:1495555200000,”regCapital”:””,”pencertileScore”:4902,”type”:1,”legalPersonName”:”温旭颖”,”toco”:2,”legalPersonId”:2051255554,”name”:”陕西蜂窝科技股份有限公司”,”logo”:””,”alias”:”蜂窝科技”,”id”:3053414776,”category”:”723″,”personType”:1,”base”:”han”},{“regStatus”:”注销”,”estiblishTime”:1473264000000,”regCapital”:””,”pencertileScore”:3860,”type”:1,”legalPersonName”:”常青”,”toco”:8,”legalPersonId”:1911055314,”name”:”陕西蜂窝科技股份有限公司”,”logo”:””,”alias”:”蜂窝科技”,”id”:2958332903,”category”:”721″,”personType”:1,”base”:”xj”}],”total”:18}}

   2.拖拽自己实现的处理器

根据要求配置好自己的json_path的路径,这样方便获取数据.如: $.result.items[*]

自己处理器的主要代码是:

@Tags({"first-example:fetch value from json string"})
@SideEffectFree
@CapabilityDescription("fetch value from json string.")
public class FirstProcessor extends AbstractProcessor {
   
   private List<PropertyDescriptor> properties;
   
   private Set<Relationship> relationships;
   
   private final String arrayFlag="true";
   
   /**
    * json路径.
    */
   public static final PropertyDescriptor JSON_PATH = new PropertyDescriptor.Builder()
         .name("Json Path")
         .required(true)
         .description("json path value,such as:$.test")
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
         .build();
   
   /**
    * json路径.
    */
   public static final PropertyDescriptor ARRAY_FLAG = new PropertyDescriptor.Builder()
         .name("Array Flag")
         .required(true)
         .description("mark if the input json is array or not")
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
         .allowableValues("true", "false")
         .defaultValue("false")
         .build();
   
   /**
    * 成功标识.
    */
   public static final Relationship SUCCESS = new Relationship.Builder()
         .name("SUCCESS")
         .description("Succes relationship")
         .build();
   
   
   @Override
   public Set<Relationship> getRelationships(){
      return relationships;
   }
   
   @Override
   public List<PropertyDescriptor> getSupportedPropertyDescriptors(){
      return properties;
   }
   
   /**
    * 初始化配置
    * @param context
    */
   @Override
   public void init(final ProcessorInitializationContext context){
      List<PropertyDescriptor> properties = new ArrayList<>();
      properties.add(JSON_PATH);
      properties.add(ARRAY_FLAG);
      this.properties = Collections.unmodifiableList(properties);
      
      Set<Relationship> relationships = new HashSet<>();
      relationships.add(SUCCESS);
      this.relationships = Collections.unmodifiableSet(relationships);
   }
   
   @Override
   public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
      final AtomicReference<String> value = new AtomicReference<>();
      FlowFile flowfile = session.get();
      session.read(flowfile, new InputStreamCallback() {
         @Override
         public void process(InputStream in) throws IOException {
            try{
               String json =IOUtils.toString(in, StandardCharsets.UTF_8.name());
               String flag = context.getProperty(ARRAY_FLAG).getValue();
               if (flag.equalsIgnoreCase(arrayFlag)){
                  List<Object> dataList = JsonPath.read(json, context.getProperty(JSON_PATH).getValue());
                  if (ObjectUtils.allNotNull(dataList)){
                     StringBuilder all = new StringBuilder("[");
                     int total = 0;
                     for (Object object : dataList) {
                        LinkedHashMap<String,Object> dataMap = (LinkedHashMap<String, Object>) object;
                        Set<String> keys = dataMap.keySet();
                        int count = 0;
                        StringBuilder builder = new StringBuilder("{");
                        for (String key :keys ) {
                           if (count==keys.size()-1){
                              builder.append("\""+key+"\":\""+dataMap.get(key)+"\"");
                           }else{
                              builder.append("\""+key+"\":\""+dataMap.get(key)+"\",");
                           }
                           count++;
                        }
                        if (total==dataList.size()-1){
                           builder.append("}");
                        }else {
                           builder.append("},");
                        }
                        total++;
                        all.append(builder.toString());
                        builder.reverse();
                     }
                     all.append("]");
                     value.set(all.toString());
                  }
               }else {
                  String result = JsonPath.read(json, context.getProperty(JSON_PATH).getValue());
                  value.set(result);
               }
            }catch(Exception ex){
               ex.printStackTrace();
               getLogger().error("failed to read json string.");
            }
         }
      });
      
      //Write the results to an attribute
      String results = value.get();
      if(results != null && !results.isEmpty()){
         String flag = context.getProperty(ARRAY_FLAG).getValue();
         if (flag.equalsIgnoreCase(arrayFlag)){
            Map<String,String> data=new HashMap<>(16);
            data.put(NiFiConstant.MATCH_ATTR,value.toString());
            flowfile = session.putAllAttributes(flowfile,data);
         }else {
            flowfile = session.putAttribute(flowfile, NiFiConstant.MATCH_ATTR, results);
         }
      }
      
      //To write the results back out ot flow file
      flowfile = session.write(flowfile, new OutputStreamCallback() {
         @Override
         public void process(OutputStream out) throws IOException {
            out.write(value.get().getBytes());
         }
      });
      
      session.transfer(flowfile, SUCCESS);
   }

}

要指明是不是一个json array ,因为jsonobject 和jsonarray 的解析和接收对象是不一样的。

3.使用PutFile

指定处理完成之后,文件的输出地址:

4.整个的流程图如下:

直接上代码,按照图所示来操作,就可以看见对应的文件输出到目录里了。

 

二、合并文本的内容

假定我们需要在把一个文本内容拼接上另外一个文本内容,那么在nifi的架子上,该如何操作呐?共分为三步:

1.Json文档的输入端

需要一个GenerateFlowFile处理器,做如下配置:

Custom text 存放的是操作的文本内容,如下所示: 你是哪个 ?

2.拖拽自己实现的处理器

根据要求配置好自己的input value的值,这样就可以将a中的文本内容进行拼接:

NiFi 学习 —自己实现处理器

代码实现如下:

@Tags({"second-example:Combine two sentences!"})
@SeeAlso({})
@SideEffectFree
@CapabilityDescription("merge two content to one together")
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})
public class SecondProcessor extends AbstractProcessor {
   
   /**
    * 属性描述对象集合
    */
   private List<PropertyDescriptor> descriptors;
   /**
    * 关联关系集合
    */
   private Set<Relationship> relationships;
   /**
    * 文件设置.
    */
   private static final String FILE_NAME = "out-";
   private static final String FILE_SUFFIX = ".txt";
   
   public static final PropertyDescriptor INPUT_VALUE = new PropertyDescriptor.Builder()
         .name("INPUT_VALUE")
         .displayName("INPUT VALUE")
         .description("input value for operating")
         .required(true)
         //非空验证
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
         .build();
   
   public static final Relationship RELATIONSHIP_SUCCESS = new Relationship.Builder()
         .name("sucess")
         .description("example relationship uccess")
         .build();
   
   public static final Relationship RELATIONSHIP_FAILURE = new Relationship.Builder()
         .name("failure")
         .description("example relationship failure")
         .build();
   
   public static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
         .name("character-set")
         .displayName("Character Set")
         .required(true)
         .defaultValue("UTF-8")
         .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
         .build();
   
   @Override
   protected void init(final ProcessorInitializationContext context) {
      final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
      descriptors.add(INPUT_VALUE);
      descriptors.add(CHARSET);
      this.descriptors = Collections.unmodifiableList(descriptors);
      
      final Set<Relationship> relationships = new HashSet<Relationship>();
      relationships.add(RELATIONSHIP_SUCCESS);
      relationships.add(RELATIONSHIP_FAILURE);
      this.relationships = Collections.unmodifiableSet(relationships);
   }
   
   @Override
   public Set<Relationship> getRelationships() {
      return this.relationships;
   }
   
   @Override
   public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
      return descriptors;
   }
   
   @OnScheduled
   public void onScheduled(final ProcessContext context) {
      getLogger().info("Processor-Name"+context.getName());
      Map<PropertyDescriptor, String> dataMap = context.getProperties();
      for (Map.Entry<PropertyDescriptor, String> entry : dataMap.entrySet()) {
         getLogger().info("key="+entry.getKey().toString()+",value="+entry.getValue());
      }
   }
   
   @Override
   public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
      FlowFile flowFile = session.get();
      if ( flowFile == null ) {
         return;
      }
      final AtomicReference<String> value = new AtomicReference<>();
      session.read(flowFile, new InputStreamCallback() {
         @Override
         public void process(InputStream inputStream) throws IOException {
            try{
               String inputVal = IOUtils.toString(inputStream, StandardCharsets.UTF_8.name());
               //utf8 的设置
               final Charset charset = Charset.forName(context.getProperty(CHARSET).getValue());
               getLogger().info("得到字符集结果是:"+charset.name());
               String current = new String(context.getProperty(INPUT_VALUE).getValue().getBytes(charset),StandardCharsets.UTF_8.name());
               String result = "处理结果:" + inputVal + current;
               //以 utf8 的方式把流信息写出去.
               getLogger().info("处理得到的结果是:"+result);
               value.set(result);
            }catch(Exception ex){
               ex.printStackTrace();
               getLogger().error("failed to read input string!");
            }
         }
      });
      
      String results = value.get();
      if(results != null && !results.isEmpty()){
         flowFile = session.putAttribute(flowFile, NiFiConstant.MATCH_ATTR, results);
      }
      
      //写入文件信息.
      flowFile = session.write(flowFile, new OutputStreamCallback() {
         @Override
         public void process(OutputStream outputStream) throws IOException {
            getLogger().info("写出的消息是:"+value.get());
            byte[] content = value.get().getBytes();
            //远程的输出流
            outputStream.write(content);
            
            //重新定义本地输出流.
            outputStream = new FileOutputStream(new File(FILE_NAME+uuid()+FILE_SUFFIX));
            outputStream.write(content);
         }
      });
      session.transfer(flowFile, RELATIONSHIP_SUCCESS);
   }
   
   /**
    * 产生一个32位的GUID
    * @return
    */
   public String uuid() {
      return getIdentifier().replace("-", "").toUpperCase();
   }
}

3.使用PutFile

指定处理完成之后,文件的输出地址:.

4.整体的流程图如下所示:

如此就可以完成两个文本内容的拼接并输出的操作。

三、给一个文件内容添加头信息

假定我们需要在把一个文件内部的内容拼接上另外一个文本内容,那么在nifi的架子上,该如何操作呐?共分为三步:

1.Json文档的输入端

需要一个GenerateFlowFile处理器,做如下配置:

注意file size 和字符集的设置。Custom text 存放的是操作的文本内容,如下所示:你是谁?

2.拖拽自己实现的处理器

根据要求配置好自己的文件的绝对路径的值,这样就可以将a中的文本内容进行拼接

需要一个绝对的文件路径,具体代码如下:

@Tags({"third-example:deal with content!"})
@SeeAlso({})
@SideEffectFree
@CapabilityDescription("add prefix to given content.")
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})
public class ThirdProcessor extends AbstractProcessor {
   
   /**
    * 属性描述对象集合
    */
   private List<PropertyDescriptor> descriptors;
   /**
    * 关联关系集合
    */
   private Set<Relationship> relationships;
   /**
    * 文件设置.
    */
   private static final String FILE_NAME = "combine-";
   private static final String FILE_SUFFIX = ".txt";
   
   public static final PropertyDescriptor ABSOLUTE_PATH = new PropertyDescriptor.Builder()
         .name("ABSOLUTE_PATH")
         .displayName("ABSOLUT PATH")
         .description("input file path for operating")
         .required(true)
         //非空验证
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
         .build();
   
   public static final Relationship SHIP_SUCCESS = new Relationship.Builder()
         .name("sucess")
         .description("example relationship uccess")
         .build();
   
   public static final Relationship SHIP_FAILURE = new Relationship.Builder()
         .name("failure")
         .description("example relationship failure")
         .build();
   
   public static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
         .name("character-set")
         .displayName("Character Set")
         .required(true)
         .defaultValue("UTF-8")
         .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
         .build();
   
   @Override
   protected void init(final ProcessorInitializationContext context) {
      final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
      descriptors.add(ABSOLUTE_PATH);
      descriptors.add(CHARSET);
      this.descriptors = Collections.unmodifiableList(descriptors);
      
      final Set<Relationship> ships = new HashSet<Relationship>();
      ships.add(SHIP_SUCCESS);
      ships.add(SHIP_FAILURE);
      this.relationships = Collections.unmodifiableSet(ships);
   }
   
   @Override
   public Set<Relationship> getRelationships() {
      return this.relationships;
   }
   
   @Override
   public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
      return descriptors;
   }
   
   @OnScheduled
   public void onScheduled(final ProcessContext context) {
      getLogger().info("Processor-Name"+context.getName());
      Map<PropertyDescriptor, String> dataMap = context.getProperties();
      for (Map.Entry<PropertyDescriptor, String> entry : dataMap.entrySet()) {
         getLogger().info("key="+entry.getKey().toString()+",value="+entry.getValue());
      }
   }
   
   @Override
   public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
      FlowFile flowFile = session.get();
      if ( flowFile == null ) {
         return;
      }
      final AtomicReference<String> value = new AtomicReference<>();
      //utf8 的设置
      final Charset charset = Charset.forName(context.getProperty(CHARSET).getValue());
      session.read(flowFile, new InputStreamCallback() {
         @Override
         public void process(InputStream inputStream) throws IOException {
            try{
               String headerDesc = IOUtils.toString(inputStream, StandardCharsets.UTF_8.name());
               String filePath = context.getProperty(ABSOLUTE_PATH).getValue();
               InputStreamReader inRd = new InputStreamReader(new FileInputStream(filePath),charset);
               BufferedReader reader = new BufferedReader(inRd);
               String line=null;
               StringBuilder  builder = new StringBuilder();
               while (null!=(line=reader.readLine())){
                  getLogger().info("文件信息是:"+line);
                  builder.append(headerDesc+new String(line.getBytes(charset),StandardCharsets.UTF_8.name())+"\n\t");
               }
               //以 utf8 的方式把流信息写出去.
               getLogger().info("处理得到的结果是:"+builder.toString());
               value.set(builder.toString());
            }catch(Exception ex){
               ex.printStackTrace();
               getLogger().error("failed to read input string!");
            }
         }
      });
      String results = value.get();
      if(results != null && !results.isEmpty()){
         flowFile = session.putAttribute(flowFile, NiFiConstant.MATCH_ATTR, results);
      }
      
      //写入文件信息.
      flowFile = session.write(flowFile, new OutputStreamCallback() {
         @Override
         public void process(OutputStream outputStream) throws IOException {
            getLogger().info("写出的消息是:"+value.get());
            byte[] content = value.get().getBytes();
            //远程的输出流
            outputStream.write(content);
            
            //重新定义本地输出流.
            outputStream = new FileOutputStream(new File(FILE_NAME+uuid()+FILE_SUFFIX));
            outputStream.write(content);
         }
      });
      session.transfer(flowFile, SHIP_SUCCESS);
   }
   
   /**
    * 产生一个32位的GUID
    * @return
    */
   public String uuid() {
      return getIdentifier().replace("-", "").toUpperCase();
   }
}

3.使用PutFile

指定处理完成之后,文件的输出地址

4.整体的流程图如下所示

如此挨个执行,不报错的情况下,就可以看见执行的结果了。

本文只是一个简单描述下如何基于nifi框架,来实现自己的业务逻辑,下一篇我就复杂使用下,看看如何操作。

如有不明白的,请微信搜索公众号 :codingba ,我会一一解答。


开心洋葱 , 版权所有丨如未注明 , 均为原创丨未经授权请勿修改 , 转载请注明NiFi 学习 —自己实现处理器
喜欢 (0)

您必须 登录 才能发表评论!

加载中……