LookupAttribute

NIFI实战 2020-12-07

1711

官网介绍

Lookup attributes from a lookup service

个人解读

一个用动态属性，通过多种多样service服务提取数据，写入flowfile属性的处理器

配置详情

Lookup Service

ElasticSearchStringLookupService(附录二)
PropertiesFileLookupService(附录一)
SimpleKeyValueLookupService(附录五)
SimpleScriptedLookupService(附录六)
XMLFileLookupService(附录七)
CouchbaseKeyValueLookupService(附录九)
SimpleCsvFileLookupService(附录三)
SimpleDatabaseLookupService(附录四)

DistributedMapCacheLookupService(附录八)

提取数据服务，详细见附录

Include Empty Values

true
false

flase情况，属性值不存在，忽略写出

动态属性

The name of the attribute to add to the FlowFile

写入flowfile的属性名称

从service里边提取数据的key

附录一：PropertiesFileLookupService

属性配置

Configuration File

properties的文件保存路径

逻辑流程图

依赖

<dependency>
  <groupId>org.apache.commons</groupId>
  <artifactId>commons-configuration2</artifactId>
  <version>2.4</version>
</dependency>
<dependency>
  <groupId>commons-beanutils</groupId>
  <artifactId>commons-beanutils</artifactId>
  <version>1.9.3</version>
</dependency>

核心代码模拟演示

new Runnable() {
  @Override
  public void run() {
    final ReloadingFileBasedConfigurationBuilder<PropertiesConfiguration> builder;
    final String config = "E:/test.properties";
    final FileBasedBuilderParameters params = new Parameters().fileBased().setFile(new File(config));
    builder = new ReloadingFileBasedConfigurationBuilder<>(PropertiesConfiguration.class).configure(params);
    builder.addEventListener(ConfigurationBuilderEvent.CONFIGURATION_REQUEST,
        new EventListener<ConfigurationBuilderEvent>() {
          @Override
          public void onEvent(ConfigurationBuilderEvent event) {
            if (builder.getReloadingController().checkForReloading(null)) {
              System.out.println("Reloading " + config);
            }
          }
        });
    try {
      builder.getConfiguration();
    } catch (ConfigurationException e) {
      System.out.println(e.getMessage());
    }
    while (true) {
      try {
        System.out.println("----------------------------");
        Configuration c = builder.getConfiguration();
        Iterator<String> it = c.getKeys();
        while (it.hasNext()) {
          String key = it.next();
          System.out.println(
              "key:" + key + ",value:" + c.getString(key)
          );
        }
        Thread.sleep(1000);
      } catch (Exception e) {
        System.out.println("ERROR");
      }
    }
  }
}.run();

结果展示

PropertiesFileLookupService一个文件配置映射服务，服务初始化要关联一个本地properties，即可形成内存的map和properties关联，properties更新立即刷入内存map，无需重启

附录二：ElasticSearchStringLookupService

属性配置

Client Service	es服务
Index	索引
Type	类型

主要功能提供es的id，通过这个id从es里边查数据。

例如LookupAttribute的每一个配置的动态属性的value，填的是es的id。

这样LookupAttribute就可以遍历动态属性，把动态属性的key通过map的形式传递给ElasticSearchStringLookupService的lookup方法，而达到查询ES的功能。

核心代码

public Optional<String> lookup(Map<String, Object> coordinates) throws LookupFailureException {
  try {
    final String id = (String) coordinates.get(ID);
    final Map<String, Object> enums = esClient.get(index, type, id);
    if (enums == null) {
      return Optional.empty();
    } else {
      return Optional.ofNullable(mapper.writeValueAsString(enums));
    }
  } catch (IOException e) {
    throw new LookupFailureException(e);
  }
}
public Set<String> getRequiredKeys() {
  return Collections.singleton(ID);
}

附录三：SimpleCsvFileLookupService

属性配置

CSV File	csv文件路径
CSV Format	Excel MySQL TDF PostgreSQLCsv InformixUnloadCsv Oracle Default RFC4180 InformixUnload PostgreSQLText csv文件类型
Character Set	文件编码类型
Lookup Key Column	查看的key
Lookup Value Column	查看的value
Ignore Duplicates	true false 是否通过key去重

功能介绍

第一步程序会把，csv的第一行解析作为head，根据配置的key和value的列，逐行写入Map集合。

LookupAttribute动态配置的value会和Map结合的key做匹配，提取数据；

如上csv数据，lgnore Duplicates是true，如果LookupAttribute配置的动态属性为（a，2）则会提取出aa值，原来的flowfile会新增属性a值是aa（a，aa）。

而当lgnore Duplicates是false的时候，csv初始化失败，SimpleCsvFileLookupService不可用。

附录四：SimpleDatabaseLookupService

配置详情

Database Connection Pooling Service	DBCPConnectionPool DBCPConnectionPoolLookup DBCPConnectionPool HiveConnectionPool DBCPConnectionPoolLookup 数据库连接池
Table Name	表名
Lookup Key Column	sql查询，条件字段
Lookup Value Column	sql查询，返回字段
Cache Size	缓存大小
Clear Cache on Enabled	是否每次启停，清理缓存
Cache Expiration	缓存有效期

功能介绍

"SELECT " + Lookup Value Column + " FROM " + Table Name+ " WHERE " + Lookup Key Column+ " = ?"

通过配置构造sql，查询数据库，并缓存数据。。

举例：LookupAttribute动态配置（key=test,mid=1）。

SimpleDatabaseLookupService配置(Table Name-test,Lookup Key Column-id,Lookup Value Column-name)生成sql为：select ${,Lookup Value Column} from ${Table Name} where ${Lookup Key Column}= ${mid}。

生成cache(mid,huhuhuhr),。

LookupAttribute最终生成的属性名称test,值为huhuhuhr。

注意点：Lookup Key Column和LookupAttribute配置的属性值不一定要一致。

附录五：SimpleKeyValueLookupService

lookupValues = context.getProperties().entrySet().stream()
  .collect(Collectors.toMap(entry -> entry.getKey().getName(), entry -> context.getProperty(entry.getKey()).evaluateAttributeExpressions().getValue()));

所有的动态属性都会被写成map的一条数据

附录六：SimpleScriptedLookupService

这个有点狠了，一般不会用我觉得，不好调试，这个service可以通过groovy自己写一个脚本，groovy里边一定要有onEable和onDisable。就是一个壳子，想怎么写就怎么写，想怎么实现就怎么实现。

附录七：XMLFileLookupService

这个使用方式SimpleCsvFileLookupService一致。

区别在于。这个是xml解析，会把xml数据解析成点记法

例如<a><b>1</b></a>,会解析成a.b=1,这样的数据。

附录八：DistributedMapCacheLookupService

Distributed Cache Service	分布式缓存服务
Character Encoding	编码类型

从分布式缓存服务里边拿数据，redis/habase/java缓存

附录九：CouchbaseKeyValueLookupService

Couchbase Cluster Controller Service

CouchbaseClusterService

hbase服务

Bucket Name

桶名称

Lookup Sub-Document Path

The Sub-Document lookup path within the target JSON document

没用过，不知道怎么用，占坑

数据库

文章转载自NIFI实战，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

LookupAttribute

评论