NIFI使用

1 从Kafka接收消息，存储到数据库中。

（1） ConsumerKafka processor

（2）Execute Scripts Processor
我这里是使用JS脚本进行处理。 还有很多其他语言的脚本。

var flowFile = session.get();
if (flowFile != null) {
   var IOUtils = Java.type("org.apache.commons.io.IOUtils");
   var StreamCallback = Java.type("org.apache.nifi.processor.io.StreamCallback");
   var StandardCharsets = Java.type("java.nio.charset.StandardCharsets");
   var DateFormatUtils=Java.type("org.apache.commons.lang3.time.DateFormatUtils");

   // var dataType=flowFile.getAttribute('data_type')
   // var FLAG=flowFile.getAttribute('flag')
   var tm = null;
   try {
      flowFile = session.write(flowFile, new StreamCallback(function (inputStream, outputStream) {
         var inputText = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
         var msg = JSON.parse(inputText);
         var stationId = msg['stationId'];
         var stationName = msg['stationName'];
         var deviceId = msg['deviceId'];
         var deviceName = msg['deviceName'];
         var deviceNo = msg['deviceNo'];
 var receiveType = msg['receiveType'];
         var createAt = msg['createAt'];
         var createAtString=DateFormatUtils.format(Number(createAt),'yyyy-MM-dd HH:mm:ss');
 var obTime = msg['obTime'];
 var obDate = msg['obDate'];
         var obDateString=DateFormatUtils.format(Number(obDate),'yyyy-MM-dd HH:mm:ss');
 var order = msg['order'];
 var distance = msg['distance'];
         var channel1SignalStrength = msg['channel1SignalStrength']
 var powerVoltage = msg['powerVoltage']
 
 var sql = 'insert into "SJZT_ODS"."water_data_distance"('
 + '"station_id", "station_name", "device_id", "device_name", "device_no", "receive_type", "create_at", "ob_time", "ob_date", "order", "distance", "channel1_signal_strength", "power_voltage")'
 + 'VALUES('
 + stationId + ', \'' + stationName + '\', ' + deviceId + ', \'' + deviceName + '\', \'' + deviceNo + '\', ' + receiveType + ', \'' + createAtString + '\', \'' + obTime + '\', \'' + obDateString + '\', ' + order + ', ' + distance + ', ' + channel1SignalStrength + ', ' + powerVoltage 
 + ')';
         outputStream.write(sql.getBytes(StandardCharsets.UTF_8));
      }));

      // flowFile = session.putAttribute(flowFile, "tm",tableName);
      session.transfer(flowFile, REL_SUCCESS);
   } catch (e) {
      flowFile = session.putAttribute(flowFile, "rsvr.transfer.error", e);
      session.transfer(flowFile, REL_FAILURE);
   }
}

注意： 这里只是生成了一个sql字符串，并没有执行sql，因此需要后面的processor来执行sql语句。
（3）PutSql processor

注意：autocommit要设置为true，否则看不到数据库里面的数据的。

2 将一堆Processors移动到一个Group里面界面操作

貌似没有直接的移动操作。
（1） Ctrl + A 全选要移动的processors
（2） 点击左边的group按钮

（3）为新的Group命名
（4）好了。选中的所有的processors都移动到了自己新创建的group中了。

参考材料

[1] https://blog.csdn.net/guijianchouxyz/article/details/120340154

原文地址：https://blog.csdn.net/kaikai_sk/article/details/144375907

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！